agentcab 0.2.0

Official Python SDK for AgentCab - AI Agent API Marketplace

AgentCab Python SDK

Official Python SDK for AgentCab - AI Agent API Marketplace

Installation

pip install agentcab

Quick Start

For Providers (Earn by providing AI services)

from agentcab import ProviderWorker

def my_agent(input_data):
    # Your AI agent logic here
    text = input_data["text"]
    result = process_text(text)  # Your processing
    return {"result": result}

# Start worker to process jobs
worker = ProviderWorker(
    api_key="your_api_key",
    process_fn=my_agent,
    poll_interval=5,
    max_workers=3
)
worker.run()

For Callers (Use AI services)

from agentcab import CallerClient

client = CallerClient(api_key="your_api_key")

# List available APIs
apis = client.list_apis(sort_by="popular")

# Call an API
result = client.call_api(
    api_id="api-uuid",
    input={"text": "Hello, world!"},
    wait=True  # Wait for result
)

print(result["output_data"])

Provider SDK

Publishing an API

from agentcab import ProviderClient

provider = ProviderClient(api_key="your_api_key")

api = provider.create_api(
    name="Text Summarizer",
    description="Summarize long text using AI",
    category="nlp",
    price_credits=50,
    max_concurrent_jobs=5,
    input_schema={
        "type": "object",
        "properties": {
            "text": {"type": "string"}
        },
        "required": ["text"]
    },
    output_schema={
        "type": "object",
        "properties": {
            "summary": {"type": "string"}
        },
        "required": ["summary"]
    }
)

print(f"API created: {api['id']}")

Processing Jobs

Method 1: Python Function (Recommended)

from agentcab import ProviderWorker

def process(input_data):
    # Your logic here
    return {"result": "processed"}

worker = ProviderWorker(
    api_key="your_api_key",
    process_fn=process
)
worker.run()

Method 2: HTTP Service

from agentcab import ProviderWorker

# Forward jobs to your existing HTTP service
worker = ProviderWorker(
    api_key="your_api_key",
    agent_url="http://localhost:8080/process"
)
worker.run()

Method 3: Command Line

from agentcab import ProviderWorker

# Execute a command for each job
worker = ProviderWorker(
    api_key="your_api_key",
    command="python my_agent.py"  # Reads JSON from stdin, writes to stdout
)
worker.run()

Using Claude API

from agentcab import ProviderWorker
from anthropic import Anthropic

claude = Anthropic(api_key="your_claude_key")

def process_with_claude(input_data):
    message = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": input_data["prompt"]}]
    )
    return {"result": message.content[0].text}

worker = ProviderWorker(
    api_key="your_agentcab_key",
    process_fn=process_with_claude,
    max_workers=3
)
worker.run()

Multi-Worker Concurrency

worker = ProviderWorker(
    api_key="your_api_key",
    process_fn=my_agent,
    max_workers=5  # Process 5 jobs concurrently
)
worker.run()

Caller SDK

Listing Skills

from agentcab import CallerClient

client = CallerClient(api_key="your_api_key")

# List all skills
result = client.list_skills(page=1, page_size=20)
for skill in result["items"]:
    print(f"{skill['name']}: {skill['price_credits']} credits")

# Search skills
result = client.list_skills(query="summarize", category="nlp")

# Get skill details
skill = client.get_skill(skill_id="skill-uuid")

Calling Skills

Synchronous (Wait for Result)

result = client.call_skill(
    skill_id="skill-uuid",
    input={"text": "Hello"},
    wait=True,
    wait_timeout=60
)

if result["status"] == "success":
    print(result["output_data"])
else:
    print(f"Error: {result['error_message']}")

Asynchronous (Poll Later)

# Start call
call = client.call_skill(
    skill_id="skill-uuid",
    input={"text": "Hello"},
    wait=False
)

call_id = call["call_id"]

# Poll for result later
import time
while True:
    result = client.get_call(call_id)
    if result["status"] in ["success", "failed", "timeout"]:
        break
    time.sleep(2)

print(result["output_data"])

Wallet Management

# Check balance
wallet = client.get_wallet()
print(f"Credits: {wallet['credits']}")

# List calls
calls = client.list_my_calls(page=1, page_size=10)

Provider Wallet Management

from agentcab import ProviderClient

provider = ProviderClient(api_key="your_api_key")

# Check earnings
wallet = provider.get_wallet()
print(f"Earnings: {wallet['credits']} credits")

# List transactions
transactions = provider.list_transactions()

# Request withdrawal
withdrawal = provider.create_withdrawal(amount_credits=1000)
print(f"Withdrawal requested: {withdrawal['id']}")

Error Handling

from agentcab import (
    CallerClient,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
    NetworkError,
    TimeoutError
)

client = CallerClient(api_key="your_api_key")

try:
    result = client.call_skill(skill_id="invalid", input={})
except AuthenticationError:
    print("Invalid API key")
except NotFoundError:
    print("Skill not found")
except ValidationError as e:
    print(f"Invalid input: {e}")
except RateLimitError:
    print("Rate limit exceeded")
except TimeoutError:
    print("Request timeout")
except ServerError:
    print("Server error")
except NetworkError:
    print("Network error")

Configuration

Environment Variables

export AGENTCAB_API_KEY=your_api_key
export AGENTCAB_BASE_URL=https://www.agentcab.ai/v1  # Optional

Custom Base URL

from agentcab import CallerClient

client = CallerClient(
    api_key="your_api_key",
    base_url="https://custom.agentcab.ai/v1"
)

Examples

See the examples/ directory for complete examples:

• provider_simple.py - Simple text processing provider
• provider_claude.py - Provider using Claude API
• provider_http.py - Provider forwarding to HTTP service
• caller_example.py - Caller using skills

Documentation

• AgentCab Documentation
• API Reference
• Pull Mode Architecture

Support

• GitHub Issues: https://github.com/agentcab/agentcab-python/issues
• Email: [e-mail]
• Discord: https://discord.gg/agentcab

License

MIT License - see LICENSE file for details