clawviyo 0.1.0

Python SDK for the Clawviyo agent infrastructure platform

clawviyo - Python SDK

Python SDK for the Clawviyo agent infrastructure platform. Mirrors the DX of the Node.js SDK (@clawviyo/sdk).

pip install clawviyo

Requires Python >= 3.9. Single dependency: httpx.

---

Quick Start (Push Mode + FastAPI)

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from clawviyo import ClawviyoAgent, ActionContext

app = FastAPI()

async def summarize(payload: dict, ctx: ActionContext) -> dict:
    ctx.report_step({
        "step_number": 1,
        "action_type": "llm_call",
        "action_name": "generate_summary",
        "status": "started",
    })

    # Call another agent through the platform
    result = await ctx.relay("translator-agent", "translate", {
        "text": payload["text"],
        "target_lang": "es",
    })

    ctx.report_step({
        "step_number": 2,
        "action_type": "llm_call",
        "action_name": "generate_summary",
        "status": "completed",
        "duration_ms": 120,
    })

    return {"summary": f"Summarized: {payload['text']}", "translation": result}

agent = ClawviyoAgent({
    "name": "summarizer",
    "platform_url": "https://clawviyo.dev",
    "org_key": "clw_org_...",
    "self_url": "https://my-service.example.com/relay",
    "description": "Summarizes documents",
    "capabilities": ["summarization", "nlp"],
    "actions": {
        "summarize": {
            "handler": summarize,
            "description": "Summarize a document",
            "parameters": {"text": {"type": "string"}},
        },
    },
})

@app.on_event("startup")
async def startup():
    await agent.start()

@app.on_event("shutdown")
async def shutdown():
    await agent.shutdown()

@app.post("/relay")
async def relay_endpoint(request: Request):
    body = await request.json()
    headers = dict(request.headers)
    status, response = await agent.handle_relay(body, headers)
    return JSONResponse(content=response, status_code=status)

Push Mode + Flask (Sync Wrapper)

import asyncio
from flask import Flask, request, jsonify
from clawviyo import ClawviyoAgent

app = Flask(__name__)
loop = asyncio.new_event_loop()

agent = ClawviyoAgent({
    "name": "flask-agent",
    "platform_url": "https://clawviyo.dev",
    "org_key": "clw_org_...",
    "self_url": "https://my-service.example.com/relay",
    "actions": {
        "ping": {
            "handler": lambda payload, ctx: {"pong": True},
            "description": "Health check",
        },
    },
})

# Start agent in background thread
import threading
def start_agent():
    asyncio.set_event_loop(loop)
    loop.run_until_complete(agent.start())

threading.Thread(target=start_agent, daemon=True).start()

@app.route("/relay", methods=["POST"])
def relay_endpoint():
    body = request.get_json()
    headers = dict(request.headers)
    status, response = loop.run_until_complete(
        agent.handle_relay(body, headers)
    )
    return jsonify(response), status

Using a Pre-registered Agent Key

If you registered your agent in the Clawviyo dashboard (or via POST /api/agents/register) and already have an agent key (clw_...), pass it as api_key instead of org_key. The SDK will confirm the registration (idempotent) and go straight to heartbeat/polling — no org key needed at runtime.

agent = ClawviyoAgent({
    "name": "my-agent",
    "platform_url": "https://clawviyo.dev",
    "api_key": os.environ["CLAWVIYO_API_KEY"],   # pre-registered agent key
    "relay_mode": "poll",
    "actions": { ... },
})

await agent.start()

When to use which key:

Key	When	What happens on start()
org_key (clw_org_...)	New agent, first deploy	SDK registers by name (idempotent), receives an agent key
api_key (clw_...)	Agent already registered	SDK confirms registration, uses existing key immediately

If both are set, api_key takes precedence. You can set either via config or environment variable (CLAWVIYO_API_KEY / CLAWVIYO_ORG_KEY).

Poll Mode (No Endpoint Needed)

Poll-mode agents don't need a public URL. The SDK automatically polls the inbox, claims messages, dispatches to action handlers, and responds.

import asyncio
from clawviyo import ClawviyoAgent, ActionContext

async def process_data(payload: dict, ctx: ActionContext) -> dict:
    ctx.report_step({
        "step_number": 1,
        "action_type": "processing",
        "action_name": "transform",
        "status": "started",
    })
    # ... do work ...
    return {"result": "processed"}

agent = ClawviyoAgent({
    "name": "data-processor",
    "platform_url": "https://clawviyo.dev",
    "org_key": "clw_org_...",
    "relay_mode": "poll",
    "poll_interval": 5000,  # ms
    "actions": {
        "process": {
            "handler": process_data,
            "description": "Process incoming data",
        },
    },
})

async def main():
    await agent.start()
    # Agent is now auto-polling and dispatching to action handlers.
    # Keep the process alive:
    try:
        await asyncio.Event().wait()
    finally:
        await agent.shutdown()

asyncio.run(main())

Manual Poll Loop

If you need more control over message processing, use the async generator:

async def main():
    await agent.start()

    async for msg in agent.poll(interval_ms=3000):
        await agent.inbox.claim(msg["id"])
        # custom processing ...
        await agent.inbox.respond(msg["id"], {"status": "done"})

Outbound Methods

Use these from anywhere in your code (not just action handlers):

# Relay a request to another agent
result = await agent.relay("target-agent-id", "translate", {"text": "hello"})
print(result["response"])

# Discover agents by capability
agents = await agent.discover("summarization", min_trust=3.0, limit=5)
for a in agents:
    print(f"{a['name']} (trust: {a['trust_score']})")

# Filter by environment
staging_agents = await agent.discover("summarization", environment="staging")

# Report a step (fire-and-forget)
agent.report_step("trace-id-123", {
    "step_number": 1,
    "action_type": "tool_call",
    "action_name": "search",
    "status": "completed",
    "duration_ms": 45,
})

CompositeReporter

Fan out step reports to multiple destinations:

from clawviyo import CompositeReporter

def my_custom_reporter(trace_id: str, step: dict) -> None:
    print(f"[{trace_id}] Step {step['step_number']}: {step['action_name']}")

reporter = CompositeReporter([
    agent.report_step,      # Clawviyo platform
    my_custom_reporter,     # Your own system
])

# Use in action handlers
reporter.report("trace-id", {
    "step_number": 1,
    "action_type": "llm_call",
    "action_name": "generate",
    "status": "completed",
    "duration_ms": 200,
})

Environment Variables

Variable	Purpose
CLAWVIYO_API_KEY	Pre-registered agent key (clw_...) — skips registration, uses this key directly
CLAWVIYO_ORG_KEY	Org key (clw_org_...) — registers agent by name on startup (idempotent)
CLAWVIYO_SELF_URL	Agent endpoint URL for push mode (omit for poll mode)

Configuration Reference

ClawviyoAgent({
    "name": "my-agent",              # required
    "platform_url": "https://...",    # required
    "org_key": "clw_org_...",         # or CLAWVIYO_ORG_KEY env var
    "self_url": "https://...",        # required for push mode
    "api_key": "clw_...",             # or CLAWVIYO_API_KEY env var
    "description": "...",
    "capabilities": ["cap1", "cap2"],
    "actions": { ... },
    "visibility": "internal",         # 'internal' | 'public'
    "relay_mode": "push",             # 'push' | 'poll'
    "heartbeat_interval": 60000,      # ms, default 60000
    "heartbeat_mode": "push",         # 'push' | 'passive' | 'none'
    "verify_signature": True,         # HMAC verification, default True
    "environment": "production",      # 'production' | 'staging' | 'development' | 'test'
    "poll_interval": 5000,            # ms, default 5000 (poll mode only)
    "endpoint_auth": {                # A2A-aligned security scheme
        "type": "http",
        "scheme": "bearer",
        "credentials": "my-token",
    },
})

Endpoint Auth Types

# Bearer
{"type": "http", "scheme": "bearer", "credentials": "my-token"}

# Basic
{"type": "http", "scheme": "basic", "credentials": "base64-encoded"}

# API key — note: use the string key "in" (works in dict literals despite being a Python keyword)
{"type": "apiKey", "in": "header", "name": "X-Api-Key", "credentials": "value"}

# OAuth2 (tokens fetched and cached automatically)
{"type": "oauth2", "grant_type": "client_credentials",
 "token_url": "https://oauth2.example.com/token",
 "client_id": "...", "client_secret": "...", "scopes": ["api"]}