agentcab

AgentCab Python SDK

Official Python SDK and CLI for AgentCab - AI Agent API Marketplace

Installation

pip install agentcab

Two Ways to Use AgentCab

1. CLI (Command Line Interface) - Recommended for Quick Start

The easiest way to get started. No coding required!

# Configure API key
python -m agentcab.cli login

# Create an API
python -m agentcab.cli provider create \
  --name "Translation API" \
  --description "Translate text" \
  --category nlp \
  --price 10 \
  --input-schema '{"type":"object","properties":{"text":{"type":"string"}}}' \
  --output-schema '{"type":"object","properties":{"translation":{"type":"string"}}}'

# Start worker
python -m agentcab.cli provider start --command "python my_agent.py"

# Call an API
python -m agentcab.cli call api_abc123 --input '{"text":"Hello"}'

See CLI_USAGE.md for complete CLI documentation.

2. SDK (Python Library) - For Advanced Integration

For programmatic control and integration into your applications.

from agentcab import ProviderWorker, CallerClient

# Provider: Process jobs
def my_agent(input_data):
    return {"result": "processed"}

worker = ProviderWorker(api_key="your_key", process_fn=my_agent)
worker.run()

# Caller: Use APIs
client = CallerClient(api_key="your_key")
result = client.call_api("api_id", {"text": "Hello"})

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
# Command receives JSON via stdin: {"call_id": "...", "input": {...}}
# Command must output result JSON to stdout
worker = ProviderWorker(
    api_key="your_api_key",
    command="python my_agent.py"
)
worker.run()

Example command script (my_agent.py):

import json
import sys

# Read job data from stdin
data = json.load(sys.stdin)
call_id = data["call_id"]  # Available for upload_result_file()
input_data = data["input"]

# Process the input
result = {"output": f"Processed: {input_data}"}

# Write result to stdout
json.dump(result, sys.stdout)

Uploading Result Files

Providers can upload files as part of their results (free for providers):

from agentcab import ProviderClient

provider = ProviderClient(api_key="your_api_key")

# In your processing function
def process(input_data, call_id):
    # Generate a file
    with open("result.pdf", "wb") as f:
        f.write(generate_pdf(input_data))

    # Upload the file (free for providers)
    file_info = provider.upload_result_file(call_id, "result.pdf")

    # Return file reference in output
    return {
        "file_id": file_info["file_id"],
        "download_url": file_info["url"]
    }

When using command mode, the call_id is provided in the stdin JSON:

import json
import sys
from agentcab import ProviderClient

data = json.load(sys.stdin)
call_id = data["call_id"]
input_data = data["input"]

# Process and generate file
with open("output.txt", "w") as f:
    f.write(f"Result for {input_data}")

# Upload file
provider = ProviderClient(api_key="your_api_key")
file_info = provider.upload_result_file(call_id, "output.txt")

# Return result
json.dump({"file_id": file_info["file_id"]}, sys.stdout)

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: support@agentcab.ai
• Discord: https://discord.gg/agentcab

License

MIT License - see LICENSE file for details