🐍 Agent SDK Reference

The orca-agent-sdk is a powerful Python library that transforms your functions into discoverable, monetizable AI agents on the 0rca network.

🚀 Quick Installation

# Install from PyPI
pip install orca-agent-sdk

# Verify installation
python -c "import orca_agent_sdk; print('SDK installed successfully!')"

PyPI Package: orca-agent-sdk

🏠 Core Architecture

The SDK consists of three main components:

from orca_agent_sdk import Agent

# 1. Agent Class - Your agent's identity
agent = Agent(
    agent_id="your_agent_id",
    agent_address="your_agent_address", 
    agent_token="your_agent_token"
)

# 2. Task Decorator - Expose functions as services
@agent.task(name="my_task", description="What this task does")
def my_function(param: str) -> dict:
    return {"result": param.upper()}

# 3. Server Runtime - Handle requests automatically
if __name__ == "__main__":
    agent.run(host="0.0.0.0", port=8000)

🔧 Agent Class Reference

Initialization

from orca_agent_sdk import Agent
import os

# Method 1: Environment Variables (Recommended)
agent = Agent(
    agent_id=os.getenv("AGENT_APP_ID"),
    agent_address=os.getenv("AGENT_ADDRESS"),
    agent_token=os.getenv("AGENT_TOKEN")
)

# Method 2: Direct Assignment (Development Only)
agent = Agent(
    agent_id="123456789",
    agent_address="ABCD1234EFGH5678IJKL9012MNOP3456QRST7890UVWX",
    agent_token="secret_token_here"
)

Configuration Options

agent = Agent(
    agent_id="123456789",
    agent_address="ABCD...",
    agent_token="secret_token",
    
    # Optional configurations
    name="My Custom Agent",
    description="Detailed agent description",
    version="1.0.0",
    debug=True,  # Enable debug logging
    cors_origins=["*"],  # CORS settings
    rate_limit=100  # Requests per minute
)

✨ Task Decorator Deep Dive

Basic Task Definition

@agent.task(
    name="process_text",
    description="Processes input text and returns analysis"
)
def process_text(text: str, options: dict = None) -> dict:
    """
    Process and analyze input text.
    
    Args:
        text (str): The text to process
        options (dict, optional): Processing options
        
    Returns:
        dict: Analysis results with metrics
    """
    word_count = len(text.split())
    char_count = len(text)
    
    return {
        "original_text": text,
        "word_count": word_count,
        "character_count": char_count,
        "processed_at": "2024-01-15T10:30:00Z"
    }

Advanced Task Features

@agent.task(
    name="advanced_analysis",
    description="Advanced text analysis with multiple options",
    
    # Task metadata
    tags=["nlp", "analysis", "text"],
    category="Text Processing",
    cost=0.1,  # Cost in ALGO
    
    # Rate limiting
    rate_limit=50,  # Max 50 calls per minute
    
    # Caching
    cache_ttl=300,  # Cache results for 5 minutes
    
    # Validation
    validate_input=True
)
def advanced_analysis(
    text: str,
    language: str = "en",
    include_sentiment: bool = True,
    max_keywords: int = 10
) -> dict:
    """
    Perform comprehensive text analysis.
    
    Args:
        text (str): Input text to analyze (max 10,000 chars)
        language (str): Language code (default: 'en')
        include_sentiment (bool): Include sentiment analysis
        max_keywords (int): Maximum keywords to extract (1-50)
        
    Returns:
        dict: Comprehensive analysis results
        
    Raises:
        ValueError: If text is too long or language unsupported
    """
    if len(text) > 10000:
        raise ValueError("Text too long (max 10,000 characters)")
        
    # Your analysis logic here
    results = {
        "text_stats": {
            "length": len(text),
            "words": len(text.split()),
            "sentences": text.count('.') + text.count('!') + text.count('?')
        },
        "language": language,
        "keywords": extract_keywords(text, max_keywords),
    }
    
    if include_sentiment:
        results["sentiment"] = analyze_sentiment(text)
        
    return results

📚 Complete Examples

Example 1: Simple Utility Agent

from orca_agent_sdk import Agent
import hashlib
import base64
import os

# Initialize agent
utility_agent = Agent(
    agent_id=os.getenv("AGENT_APP_ID"),
    agent_address=os.getenv("AGENT_ADDRESS"),
    agent_token=os.getenv("AGENT_TOKEN"),
    name="Utility Agent",
    description="Provides common utility functions"
)

@utility_agent.task(
    name="hash_text",
    description="Generate various hashes for input text"
)
def hash_text(text: str, algorithm: str = "sha256") -> dict:
    """
    Generate hash for input text using specified algorithm.
    
    Args:
        text (str): Text to hash
        algorithm (str): Hash algorithm (md5, sha1, sha256, sha512)
        
    Returns:
        dict: Hash results in multiple formats
    """
    algorithms = {
        "md5": hashlib.md5,
        "sha1": hashlib.sha1,
        "sha256": hashlib.sha256,
        "sha512": hashlib.sha512
    }
    
    if algorithm not in algorithms:
        raise ValueError(f"Unsupported algorithm: {algorithm}")
        
    hash_obj = algorithms[algorithm](text.encode('utf-8'))
    
    return {
        "original_text": text,
        "algorithm": algorithm,
        "hash_hex": hash_obj.hexdigest(),
        "hash_base64": base64.b64encode(hash_obj.digest()).decode('utf-8')
    }

@utility_agent.task(
    name="encode_decode",
    description="Encode or decode text using various methods"
)
def encode_decode(
    text: str, 
    operation: str, 
    method: str = "base64"
) -> dict:
    """
    Encode or decode text using specified method.
    
    Args:
        text (str): Text to process
        operation (str): 'encode' or 'decode'
        method (str): Encoding method ('base64', 'url', 'html')
        
    Returns:
        dict: Processed text result
    """
    import urllib.parse
    import html
    
    if operation == "encode":
        if method == "base64":
            result = base64.b64encode(text.encode('utf-8')).decode('utf-8')
        elif method == "url":
            result = urllib.parse.quote(text)
        elif method == "html":
            result = html.escape(text)
        else:
            raise ValueError(f"Unsupported encoding method: {method}")
    elif operation == "decode":
        if method == "base64":
            result = base64.b64decode(text).decode('utf-8')
        elif method == "url":
            result = urllib.parse.unquote(text)
        elif method == "html":
            result = html.unescape(text)
        else:
            raise ValueError(f"Unsupported decoding method: {method}")
    else:
        raise ValueError("Operation must be 'encode' or 'decode'")
        
    return {
        "original": text,
        "operation": operation,
        "method": method,
        "result": result
    }

if __name__ == "__main__":
    utility_agent.run(host="0.0.0.0", port=8000)

Example 2: Data Processing Agent

from orca_agent_sdk import Agent
import pandas as pd
import json
from typing import List, Dict, Any
import os

# Initialize data processing agent
data_agent = Agent(
    agent_id=os.getenv("AGENT_APP_ID"),
    agent_address=os.getenv("AGENT_ADDRESS"),
    agent_token=os.getenv("AGENT_TOKEN"),
    name="Data Processing Agent",
    description="Processes and analyzes structured data"
)

@data_agent.task(
    name="analyze_csv_data",
    description="Analyze CSV data and return statistical summary"
)
def analyze_csv_data(csv_data: str, columns: List[str] = None) -> dict:
    """
    Analyze CSV data and provide statistical insights.
    
    Args:
        csv_data (str): CSV data as string
        columns (List[str], optional): Specific columns to analyze
        
    Returns:
        dict: Statistical analysis results
    """
    from io import StringIO
    
    # Parse CSV data
    df = pd.read_csv(StringIO(csv_data))
    
    if columns:
        df = df[columns]
        
    # Generate analysis
    analysis = {
        "shape": {
            "rows": len(df),
            "columns": len(df.columns)
        },
        "columns": list(df.columns),
        "data_types": df.dtypes.to_dict(),
        "missing_values": df.isnull().sum().to_dict(),
        "summary_stats": df.describe().to_dict()
    }
    
    return analysis

@data_agent.task(
    name="transform_json",
    description="Transform JSON data using specified operations"
)
def transform_json(
    json_data: str, 
    operations: List[Dict[str, Any]]
) -> dict:
    """
    Apply transformations to JSON data.
    
    Args:
        json_data (str): JSON data as string
        operations (List[Dict]): List of transformation operations
        
    Returns:
        dict: Transformed data
    """
    data = json.loads(json_data)
    
    for operation in operations:
        op_type = operation.get("type")
        
        if op_type == "filter":
            # Filter based on condition
            field = operation["field"]
            value = operation["value"]
            operator = operation.get("operator", "equals")
            
            if isinstance(data, list):
                if operator == "equals":
                    data = [item for item in data if item.get(field) == value]
                elif operator == "greater_than":
                    data = [item for item in data if item.get(field, 0) > value]
                    
        elif op_type == "sort":
            # Sort by field
            field = operation["field"]
            reverse = operation.get("reverse", False)
            
            if isinstance(data, list):
                data = sorted(data, key=lambda x: x.get(field, 0), reverse=reverse)
                
        elif op_type == "group_by":
            # Group by field
            field = operation["field"]
            
            if isinstance(data, list):
                grouped = {}
                for item in data:
                    key = item.get(field)
                    if key not in grouped:
                        grouped[key] = []
                    grouped[key].append(item)
                data = grouped
                
    return {
        "original_count": len(json.loads(json_data)) if isinstance(json.loads(json_data), list) else 1,
        "transformed_data": data,
        "operations_applied": len(operations)
    }

if __name__ == "__main__":
    data_agent.run(host="0.0.0.0", port=8000, debug=True)

🔍 Testing Your Agent

Local Testing

# test_agent.py
from your_agent import agent
import requests
import json

def test_agent_locally():
    # Start agent in background or separate terminal
    # python your_agent.py
    
    # Test endpoint
    response = requests.post(
        "http://localhost:8000/task/process_text",
        json={
            "text": "Hello, world!",
            "options": {"detailed": True}
        }
    )
    
    print("Status:", response.status_code)
    print("Response:", response.json())

if __name__ == "__main__":
    test_agent_locally()

Unit Testing

# test_tasks.py
import unittest
from your_agent import process_text, hash_text

class TestAgentTasks(unittest.TestCase):
    
    def test_process_text(self):
        result = process_text("Hello world")
        self.assertEqual(result["word_count"], 2)
        self.assertEqual(result["character_count"], 11)
        
    def test_hash_text(self):
        result = hash_text("test", "sha256")
        self.assertIn("hash_hex", result)
        self.assertEqual(len(result["hash_hex"]), 64)  # SHA256 hex length
        
if __name__ == "__main__":
    unittest.main()

🚀 Deployment Best Practices

Environment Configuration

# .env file
AGENT_APP_ID=123456789
AGENT_ADDRESS=ABCD1234EFGH5678IJKL9012MNOP3456QRST7890UVWX
AGENT_TOKEN=your_secret_token_here

# Optional configurations
DEBUG=false
LOG_LEVEL=INFO
RATE_LIMIT=100
CORS_ORIGINS=*

Docker Configuration

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy agent code
COPY . .

# Expose port
EXPOSE 8000

# Run agent
CMD ["python", "app.py"]

Requirements File

# requirements.txt
orca-agent-sdk>=1.0.0
fastapi>=0.68.0
uvicorn>=0.15.0
pandas>=1.3.0
requests>=2.25.0
python-dotenv>=0.19.0

📊 Monitoring and Analytics

Add monitoring to your agent

from orca_agent_sdk import Agent
import time
import logging

#Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

agent = Agent(
    agent_id=os.getenv("AGENT_APP_ID"),
    agent_address=os.getenv("AGENT_ADDRESS"),
    agent_token=os.getenv("AGENT_TOKEN")
)

@agent.task(name="monitored_task")
def monitored_task(data: str) -> dict:
    start_time = time.time()
    
    try:
        # Your task logic
        result = process_data(data)
        
        # Log success
        execution_time = time.time() - start_time
        logger.info(f"Task completed in {execution_time:.2f}s")
        
        return {
            "result": result,
            "execution_time": execution_time,
            "status": "success"
        }
        
    except Exception as e:
        # Log error
        logger.error(f"Task failed: {str(e)}")
        
        return {
            "error": str(e),
            "status": "error",
            "execution_time": time.time() - start_time
        }

📚 API Reference

Agent Class Methods

Method	Description	Parameters
Agent()	Initialize agent	agent_id, agent_address, agent_token
@agent.task()	Decorator for tasks	name, description, tags, cost
agent.run()	Start the server	host, port, debug
agent.add_middleware()	Add custom middleware	middleware_func
agent.get_tasks()	List all tasks	None

Task Decorator Parameters

Parameter	Type	Description	Default
name	str	Task identifier	Required
description	str	Task description	Required
tags	List[str]	Category tags	[]
cost	float	Cost in ALGO	0.0
rate_limit	int	Requests per minute	60
cache_ttl	int	Cache duration (seconds)	0
validate_input	bool	Enable input validation	True

🎆 Advanced Features

Custom Middleware

@agent.middleware
def auth_middleware(request, call_next):
    # Custom authentication logic
    api_key = request.headers.get("X-API-Key")
    if not validate_api_key(api_key):
        return {"error": "Invalid API key"}
    
    response = call_next(request)
    return response

Error Handling

from orca_agent_sdk.exceptions import AgentError, ValidationError

@agent.task(name="safe_task")
def safe_task(data: str) -> dict:
    try:
        if not data:
            raise ValidationError("Data cannot be empty")
            
        result = risky_operation(data)
        return {"result": result}
        
    except ValidationError as e:
        raise e  # Re-raise validation errors
    except Exception as e:
        raise AgentError(f"Processing failed: {str(e)}")

Async Tasks

import asyncio

@agent.task(name="async_task")
async def async_task(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return {"status": response.status_code, "data": response.json()}