Metadata-Version: 2.4
Name: clawviyo
Version: 0.1.0
Summary: Python SDK for the Clawviyo agent infrastructure platform
Project-URL: Homepage, https://github.com/clawviyo/clawviyo
Project-URL: Documentation, https://docs.clawviyo.dev
Project-URL: Repository, https://github.com/clawviyo/clawviyo
Author-email: Clawviyo <hello@clawviyo.dev>
License-Expression: MIT
Keywords: a2a,agent,ai-agent,infrastructure,multi-agent,observability,sdk
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24
Description-Content-Type: text/markdown

# clawviyo - Python SDK

Python SDK for the [Clawviyo](https://clawviyo.dev) 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)

```python
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)

```python
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.

```python
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.

```python
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:

```python
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):

```python
# 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:

```python
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

```python
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

```python
# 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"]}
```
