Python SDK Guide

The ClawPipe Python SDK mirrors the TypeScript API with Pythonic conventions. Supports both async and sync usage.

Install

pip install clawpipe-ai

Requires Python 3.9+. For async support, use Python 3.10+ with asyncio.

Async Usage (Recommended)

import asyncio
from clawpipe import ClawPipe

async def main():
    pipe = ClawPipe(
        api_key="cp_xxx",
        project_id="my-app",
        enable_booster=True,
        enable_packer=True,
        enable_cache=True,
    )

    result = await pipe.prompt(
        "Explain recursion in simple terms",
        system="You are a helpful tutor",
        max_tokens=2000,
    )

    print(result.text)
    print(result.meta.context_savings)  # '42%'
    print(result.meta.estimated_cost_usd)  # 0.003

asyncio.run(main())

Sync Usage

from clawpipe import ClawPipeSync

pipe = ClawPipeSync(
    api_key="cp_xxx",
    project_id="my-app",
)

result = pipe.prompt("What is 2+2?")
print(result.text)  # "4" (resolved by Booster, zero cost)

Streaming

async for chunk in pipe.stream("Analyze this code", system="..."):
    print(chunk, end="", flush=True)

Configuration Options

All config options use snake_case Python conventions. They map 1:1 with the TypeScript SDK.

Option	Type	Default	Description
api_key	str	required	Your ClawPipe API key
project_id	str	required	Project identifier
gateway_url	str	https://api.clawpipe.ai/v1	Gateway endpoint
cache_ttl_ms	int	300000	Cache TTL in milliseconds
enable_booster	bool	True	Enable Agent Booster
enable_packer	bool	True	Enable Context Packer
enable_cache	bool	True	Enable Semantic Cache
enable_trace	bool	False	Enable pipeline tracing
local_model_url	str	None	URL of a local LLM server
enable_local_fallback	bool	False	Auto-detect local models
budget_cap_usd	float	None	Hard budget cap in USD
budget_warn_usd	float	None	Soft budget warning
rate_limit_per_day	int	tier default	Max calls per day
allowlist	list[dict]	None	Permitted providers/models
denylist	list[dict]	None	Blocked providers/models
enable_audit	bool	False	Enable audit logging
audit_transport	callable	None	Custom audit log handler
circuit_breaker_threshold	int	5	Failures before circuit opens
circuit_breaker_recovery_ms	int	30000	Recovery time in ms

Swarm Orchestration

from clawpipe import Swarm, Gateway

swarm = Swarm(
    models=[
        {"provider": "openai", "model": "gpt-4o", "quality_score": 0.94},
        {"provider": "anthropic", "model": "claude-sonnet-4", "quality_score": 0.95},
    ],
    strategy="vote",  # 'first' | 'vote' | 'best' | 'merge'
)

result = await swarm.run("Is this code safe?", {}, gateway)

Observability

pipe.stats()            # Telemetry snapshot
pipe.budget_status()    # Budget: spent, cap, remaining
pipe.rate_limit_status()# Rate limits: remaining, reset time
pipe.circuit_status()   # Provider health per provider
pipe.audit_logs()       # Timestamped action log

Error Handling

from clawpipe import BudgetExceededError, RateLimitError, GatewayError

try:
    result = await pipe.prompt("Hello")
except BudgetExceededError:
    print("Budget cap reached")
except RateLimitError:
    print("Rate limit exceeded")
except GatewayError as e:
    print(f"Provider error: {e}")