clawdforge/clients/python

clawdforge — Python SDK

Sync Python client for the clawdforge HTTP service. Wraps the bearer-token-gated REST API behind a typed dataclass surface so other Sulkta apps (cauldron, petalparse, johnny5) don't have to hand-roll requests calls.

Install

pip install -e clients/python/

(LAN-only, not on PyPI.)

Quickstart

from clawdforge import Forge

forge = Forge(base_url="http://192.168.0.5:8800", token="cf_...")

print(forge.healthz())
# {'ok': True, 'claude_present': True, 'claude_version': '1.x.y'}

r = forge.run(prompt='Reply with JSON: {"hello": "world"}')
print(r.result)        # {'hello': 'world'}
print(r.duration_ms)   # 4321
print(r.stop_reason)   # 'end_turn'

Forge is also a context manager — preferred when you have a script-shaped lifetime:

with Forge(base_url="...", token="...") as forge:
    forge.run(prompt="...")

Construction

Forge(
    *,
    base_url: str,
    token: str,
    default_model: str = "sonnet",
    default_timeout_secs: int = 120,
    http_timeout_margin: int = 30,
    session: requests.Session | None = None,
)

• base_url — http://host:port. Trailing slash stripped.
• token — bearer. App token (cf_...) for /run and /files; admin bootstrap token for /admin/*.
• default_model — passed to /run when caller doesn't override. Server default is also sonnet so this matches.
• default_timeout_secs — timeout_secs for /run when caller doesn't override.
• http_timeout_margin — seconds added to the run subprocess timeout to derive the HTTP-level timeout. The HTTP deadline must outlast the server's subprocess deadline so we don't bail while clawdforge is still doing legitimate work for us. Default 30s — same pattern cauldron has been running inline.
• session — bring your own requests.Session if you want shared connection-pooling across multiple Forges. If omitted, the Forge owns one and closes it on exit.

Methods

healthz() -> dict

forge.healthz()
# {'ok': True, 'claude_present': True, 'claude_version': '1.x.y'}

run(prompt, *, model=None, system=None, files=None, timeout_secs=None) -> RunResult

r = forge.run(
    prompt="Sterilize this ingredient line: 'about 2 cups of cooked white rice'",
    model="sonnet",            # optional, defaults to forge.default_model
    system="You are a precise recipe parser. Always reply with valid JSON.",
    files=["ff_..."],          # optional file tokens from upload_file()
    timeout_secs=60,           # optional, 5..600, defaults to forge.default_timeout_secs
)

r.ok            # True
r.result        # parsed JSON when claude returned valid JSON, else str
r.duration_ms   # 4321
r.stop_reason   # 'end_turn' | 'timeout' | 'error' | ...

upload_file(path_or_fileobj, *, ttl_secs=3600, filename=None, content_type=None) -> FileToken

Either path-like or a binary file-object:

ft = forge.upload_file("/path/to/recipe.png", ttl_secs=3600)
# FileToken(file_token='ff_...', ttl_secs=3600, size=12345)

# Or from a buffer:
import io
ft = forge.upload_file(io.BytesIO(b"..."), filename="snippet.txt")

r = forge.run(prompt="Extract recipe data", files=[ft.file_token])

The server enforces 60 <= ttl_secs <= 86400 and returns 400 for out-of-range values, surfaced as ForgeAPIError.

Admin (admin-bootstrap-token gated)

admin = Forge(base_url="http://192.168.0.5:8800", token=ADMIN_BOOTSTRAP_TOKEN)

t = admin.create_token("cauldron", ip_cidrs=["172.24.0.0/16"])
# AppToken(name='cauldron', token='cf_brandnew_xxx', ip_cidrs=['172.24.0.0/16'], ...)
# t.token is the plaintext bearer — store it now, the server only keeps a hash.

admin.list_tokens()
# [AppToken(name='cauldron', token=None, ip_cidrs=['172.24.0.0/16'], created_at=..., last_used=..., enabled=True), ...]

admin.revoke_token("cauldron")  # True on success; raises ForgeAPIError(404) if no such token

Errors

Everything the SDK raises descends from ForgeError:

ForgeError
├── ForgeTransportError    — connection/TCP-timeout failures (server never responded)
└── ForgeAPIError          — server returned 4xx/5xx; .status_code, .body available
    └── ForgeAuthError     — 401 or 403 specifically

ForgeAPIError carries:

• .status_code: int
• .body: dict | str | None — parsed JSON if available, else text
• .message: str — short summary

Typical pattern:

from clawdforge import Forge, ForgeAuthError, ForgeAPIError, ForgeTransportError

try:
    r = forge.run(prompt="...")
except ForgeAuthError:
    # Re-mint your token, or check the IP allowlist.
    raise
except ForgeAPIError as e:
    if e.status_code == 502 and isinstance(e.body, dict):
        # Subprocess failed/timed out. e.body has 'error', 'stderr',
        # 'duration_ms', 'stop_reason'.
        log.warning("clawdforge run failed: %s", e.body.get("error"))
    raise
except ForgeTransportError:
    # Network blip — clawdforge is down or unreachable. Caller decides
    # whether to retry; the SDK does not retry.
    raise

No retries by default

Clawdforge runs are not idempotent — each /run spawns a real claude -p subprocess that costs tokens and time. If you want retries, wrap calls with whatever retry policy fits your call-site (tenacity, backoff, hand-rolled). The SDK won't double-charge you behind your back.

Tests

cd clients/python
pip install -e .[test]
python -m unittest discover tests

Tests use responses to intercept HTTP calls — no live network, no live clawdforge.

Notes

• The result field of RunResult matches whatever the server's runner parsed out of claude -p --output-format json: a dict when the prompt asked for JSON and the model complied, a string otherwise. The SDK passes it through as-is — your prompt determines the shape.
• The server's GET /admin/tokens returns ip_cidrs as a comma-joined string (it's stored that way in SQLite). The SDK normalizes that to list[str] on AppToken. The create response already returns a list, so both shapes round-trip cleanly.
• revoke_token on a non-existent name raises ForgeAPIError(404) rather than returning False. That matches the server's contract and lets callers distinguish "didn't exist" from "was disabled".