Python SDK - MeshAI Protocol

Installation

Install the MeshAI Python SDK using pip:

pip install meshai-sdk

For development versions:

pip install meshai-sdk[dev]

Quick Start

from meshai import MeshAI

# Initialize client
client = MeshAI(api_key="your_api_key")

# Execute a simple task
result = client.execute_task(
    task_type="text_generation",
    input="Write a blog post about AI",
    quality_level="high"
)

print(result.output)

Client Configuration

Basic Configuration

import os
from meshai import MeshAI

# Using environment variables
os.environ['MESHAI_API_KEY'] = 'your_api_key'
client = MeshAI()

Core Methods

execute_task()

Execute a single AI task with automatic agent selection.

def execute_task(
    task_type: str,
    input_data: Any,
    quality_level: str = "standard",
    max_cost: float = None,
    timeout: int = 30000,
    agent_requirements: dict = None
) -> TaskResult

Parameters:

Parameter	Type	Description	Default
`task_type`	`str`	Type of AI task to execute	Required
`input_data`	`Any`	Input data for the task	Required
`quality_level`	`str`	Quality level: “basic”, “standard”, “high”, “premium"	"standard”
`max_cost`	`float`	Maximum cost in SOL	None
`timeout`	`int`	Timeout in milliseconds	30000
`agent_requirements`	`dict`	Specific agent requirements	None

Returns: TaskResult object with output, metadata, and metrics.

Basic Usage
With Options
Error Handling

# Simple text generation
result = client.execute_task(
    task_type="text_generation",
    input_data="Explain quantum computing"
)

print(f"Output: {result.output}")
print(f"Quality: {result.quality_score}")
print(f"Cost: {result.cost} SOL")

# Task with specific requirements
result = client.execute_task(
    task_type="image_analysis",
    input_data={"image_url": "https://example.com/image.jpg"},
    quality_level="high",
    max_cost=0.005,
    timeout=60000,
    agent_requirements={
        "min_reputation": 0.95,
        "specialization": "medical_imaging"
    }
)

from meshai.exceptions import (
    TaskTimeoutError,
    InsufficientFundsError,
    QualityThresholdError
)

try:
    result = client.execute_task(
        task_type="text_generation",
        input_data="Write a novel",
        quality_level="premium",
        timeout=120000
    )
except TaskTimeoutError:
    print("Task took too long to complete")
except InsufficientFundsError as e:
    print(f"Need {e.required_amount} SOL")
except QualityThresholdError as e:
    print(f"Quality {e.actual_quality} below threshold")

create_workflow()

Create multi-step workflows with task dependencies.

def create_workflow(
    name: str = None,
    description: str = None,
    max_parallel: int = 5
) -> Workflow

Example:

# Create workflow
workflow = client.create_workflow(name="document_analysis")

# Add OCR task
ocr_task = workflow.add_task(
    task_type="document_ocr",
    input_data={"document_url": "https://example.com/doc.pdf"},
    quality_threshold=0.99
)

# Add analysis task (depends on OCR)
analysis_task = workflow.add_task(
    task_type="document_analysis",
    input_data=ocr_task.output,
    depends_on=ocr_task
)

# Execute workflow
results = await workflow.execute()

Workflow Management

Workflow Class

class Workflow:
    def add_task(self, task_type: str, input_data: Any = None, **kwargs) -> Task
    def remove_task(self, task_id: str) -> bool
    def execute(self, max_parallel: int = 5) -> WorkflowResult
    def get_status(self) -> WorkflowStatus
    def cancel(self) -> bool

Task Dependencies

Sequential Tasks
Parallel Tasks
Complex Dependencies

workflow = client.create_workflow()

# Task 1
task1 = workflow.add_task("ocr", input_data=document)

# Task 2 depends on Task 1
task2 = workflow.add_task(
    "text_analysis",
    input_data=task1.output,
    depends_on=task1
)

# Task 3 depends on Task 2
task3 = workflow.add_task(
    "summarization",
    input_data=task2.output,
    depends_on=task2
)

workflow = client.create_workflow()

# Base task
ocr_task = workflow.add_task("ocr", input_data=document)

# Parallel tasks (both depend on OCR)
sentiment_task = workflow.add_task(
    "sentiment_analysis",
    input_data=ocr_task.output,
    depends_on=ocr_task
)

entity_task = workflow.add_task(
    "entity_extraction", 
    input_data=ocr_task.output,
    depends_on=ocr_task,
    parallel_to=sentiment_task
)

workflow = client.create_workflow()

# Multiple input tasks
ocr_task = workflow.add_task("ocr", input_data=document)
image_task = workflow.add_task("image_analysis", input_data=image)

# Task depending on multiple inputs
synthesis_task = workflow.add_task(
    "content_synthesis",
    input_data={
        "text": ocr_task.output,
        "image_analysis": image_task.output
    },
    depends_on=[ocr_task, image_task]
)

Task Types

Available Task Types

Task Type	Description	Input Format	Output Format
`text_generation`	Generate text content	`str` or `dict`	`str`
`text_analysis`	Analyze text sentiment, entities	`str`	`dict`
`text_summarization`	Summarize long text	`str`	`str`
`document_ocr`	Extract text from documents	`dict` with URL/base64	`str`
`image_analysis`	Analyze and caption images	`dict` with URL/base64	`dict`
`image_generation`	Generate images from text	`str`	`dict` with URL
`code_generation`	Generate code	`str` or `dict`	`str`
`translation`	Translate text	`dict` with text and languages	`str`
`audio_transcription`	Convert speech to text	`dict` with audio URL	`str`

Task-Specific Examples

Text Generation
Image Analysis
Translation

result = client.execute_task(
    task_type="text_generation",
    input_data={
        "prompt": "Write a technical blog post about blockchain",
        "max_tokens": 2000,
        "style": "professional",
        "audience": "developers"
    }
)

result = client.execute_task(
    task_type="image_analysis",
    input_data={
        "image_url": "https://example.com/image.jpg",
        "analysis_type": ["objects", "text", "faces"],
        "detail_level": "high"
    }
)

result = client.execute_task(
    task_type="translation",
    input_data={
        "text": "Hello, how are you?",
        "source_language": "en",
        "target_language": "es",
        "formality": "formal"
    }
)

Response Objects

TaskResult

class TaskResult:
    output: Any              # Task output data
    quality_score: float     # Quality score (0-1)
    cost: float             # Cost in SOL
    agent_id: str           # Processing agent ID
    latency: int            # Processing time (ms)
    metadata: dict          # Additional metadata
    timestamp: datetime     # Completion timestamp

WorkflowResult

class WorkflowResult:
    results: Dict[str, TaskResult]  # Results by task ID
    total_cost: float              # Total workflow cost
    total_latency: int             # Total execution time
    success_rate: float            # Percentage of successful tasks
    metadata: dict                 # Workflow metadata

Error Handling

Exception Types

from meshai.exceptions import (
    MeshAIError,           # Base exception
    TaskTimeoutError,      # Task exceeded timeout
    InsufficientFundsError,# Not enough funds
    QualityThresholdError, # Quality below threshold
    AgentUnavailableError, # No agents available
    ValidationError,       # Input validation failed
    NetworkError,          # Network connectivity issues
    AuthenticationError    # Invalid API key
)

Error Handling Patterns

Basic Error Handling
Retry Logic
Graceful Degradation

try:
    result = client.execute_task(
        task_type="text_generation",
        input_data="Generate content"
    )
except MeshAIError as e:
    print(f"MeshAI error: {e}")
    # Handle specific error
    if e.code == "INSUFFICIENT_FUNDS":
        print(f"Need {e.required_amount} SOL")
except Exception as e:
    print(f"Unexpected error: {e}")

import time

def execute_with_retry(client, task_type, input_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.execute_task(task_type, input_data)
        except (NetworkError, AgentUnavailableError) as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed, retrying in {wait_time}s")
                time.sleep(wait_time)
            else:
                raise e

def generate_content_with_fallback(prompt):
    try:
        # Try premium quality first
        result = client.execute_task(
            task_type="text_generation",
            input_data=prompt,
            quality_level="premium"
        )
        return result.output
    except InsufficientFundsError:
        # Fallback to standard quality
        result = client.execute_task(
            task_type="text_generation", 
            input_data=prompt,
            quality_level="standard"
        )
        return result.output
    except AgentUnavailableError:
        # Final fallback
        return "Content generation temporarily unavailable"

Advanced Features

Batch Processing

# Process multiple tasks efficiently
tasks = [
    {"task_type": "text_analysis", "input": text1},
    {"task_type": "text_analysis", "input": text2}, 
    {"task_type": "text_analysis", "input": text3}
]

batch = client.create_batch(tasks)
results = await batch.execute()

for i, result in enumerate(results):
    print(f"Task {i+1}: {result.output}")

Streaming Results

# Stream results for long-running tasks
async for chunk in client.stream_task(
    task_type="text_generation",
    input_data="Write a long article about AI"
):
    print(chunk.content, end="", flush=True)

Custom Agent Selection

# Target specific agents
result = client.execute_task(
    task_type="text_generation",
    input_data="Technical documentation",
    agent_requirements={
        "agent_id": "technical-writer-v2",
        "min_reputation": 0.95,
        "max_cost": 0.01,
        "geographic_preference": "us-east"
    }
)

Monitoring and Analytics

Usage Statistics

# Get usage statistics
stats = client.get_usage_stats(period="last_30_days")

print(f"Total tasks: {stats.total_tasks}")
print(f"Total cost: {stats.total_cost} SOL") 
print(f"Average quality: {stats.average_quality}")
print(f"Success rate: {stats.success_rate}")

Task History

# Get detailed task history
history = client.get_task_history(
    limit=100,
    task_type="text_generation",
    start_date="2024-01-01"
)

for task in history:
    print(f"{task.timestamp}: {task.task_type} - {task.status}")

Performance Monitoring

# Monitor real-time performance
monitor = client.create_monitor()

@monitor.on_task_complete
def handle_task_complete(result):
    if result.quality_score < 0.8:
        print(f"Low quality detected: {result.quality_score}")
    
@monitor.on_error
def handle_error(error):
    print(f"Task failed: {error}")

# Start monitoring
monitor.start()

Configuration Options

Environment Variables

# API Configuration
MESHAI_API_KEY=your_api_key
MESHAI_NETWORK=mainnet
MESHAI_TIMEOUT=30000

# Performance Tuning
MESHAI_MAX_RETRIES=3
MESHAI_QUALITY_THRESHOLD=0.8
MESHAI_MAX_COST=0.1

# Regional Preferences  
MESHAI_PREFERRED_REGIONS=us-east,eu-west
MESHAI_ENABLE_CACHING=true

Configuration File

# meshai_config.py
config = {
    "api_key": "your_api_key",
    "network": "mainnet",
    "timeout": 30000,
    "retry_attempts": 3,
    "quality_threshold": 0.9,
    "max_cost": 0.1,
    "preferred_regions": ["us-east", "eu-west"],
    "enable_caching": True,
    "log_level": "info",
    "agent_preferences": {
        "text_generation": {
            "min_reputation": 0.9,
            "max_response_time": 5000
        }
    }
}

# Load configuration
client = MeshAI(config=config)

Testing and Development

Mock Client for Testing

from meshai.testing import MockMeshAI

# Use mock client for testing
client = MockMeshAI()

# Mock responses
client.mock_response(
    task_type="text_generation",
    response=TaskResult(
        output="Mocked response",
        quality_score=0.95,
        cost=0.001
    )
)

# Test your code
result = client.execute_task("text_generation", "test input")
assert result.output == "Mocked response"

Development Mode

# Enable development mode
client = MeshAI(
    api_key="test_key",
    network="testnet",
    development_mode=True,
    verbose_logging=True
)

Best Practices

Error Handling

Always implement proper error handling for production applications:

Handle network timeouts gracefully
Implement retry logic for transient failures
Provide fallback mechanisms for critical features
Log errors for debugging and monitoring

Performance Optimization

Optimize your usage for better performance and costs:

Use appropriate quality levels for your use case
Implement caching for repeated similar requests
Use batch processing for multiple related tasks
Monitor usage and costs regularly

Security

Follow security best practices:

Store API keys securely (environment variables)
Validate and sanitize input data
Use HTTPS for all communications
Monitor for unusual usage patterns

Next Steps

JavaScript SDK

Explore the JavaScript SDK for web and Node.js applications

Examples

See real-world implementation examples and use cases

Agent Development

Learn how to create and monetize your own AI agents

Community

Join the developer community for support and discussions

SDK Documentation

Examples

​Installation

​Quick Start

​Client Configuration

​Basic Configuration

​Core Methods

​execute_task()

​create_workflow()

​Workflow Management

​Workflow Class

​Task Dependencies

​Task Types

​Available Task Types

​Task-Specific Examples

​Response Objects

​TaskResult

​WorkflowResult

​Error Handling

​Exception Types

​Error Handling Patterns

​Advanced Features

​Batch Processing

​Streaming Results

​Custom Agent Selection

​Monitoring and Analytics

​Usage Statistics

​Task History

​Performance Monitoring

​Configuration Options

​Environment Variables

​Configuration File

​Testing and Development

​Mock Client for Testing

​Development Mode

​Best Practices

​Next Steps

JavaScript SDK

Examples

Agent Development

Community

Installation

Quick Start

Client Configuration

Basic Configuration

Core Methods

execute_task()

create_workflow()

Workflow Management

Workflow Class

Task Dependencies

Task Types

Available Task Types

Task-Specific Examples

Response Objects

TaskResult

WorkflowResult

Error Handling

Exception Types

Error Handling Patterns

Advanced Features

Batch Processing

Streaming Results

Custom Agent Selection

Monitoring and Analytics

Usage Statistics

Task History

Performance Monitoring

Configuration Options

Environment Variables

Configuration File

Testing and Development

Mock Client for Testing

Development Mode

Best Practices

Next Steps