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="...")

Multi-turn / Sessions (v0.2)

For agent-shaped workflows that span multiple turns, use a Session. A session keeps acpx context across turn() calls — file reads, tool-call results, and conversational state persist across turns until the session is closed (or hits the server's TTL).

Block form (preferred — auto-close)

with forge.session(agent="claude") as s:
    r1 = s.turn("Read README.md and summarize it")
    print(r1.text())          # concatenated 'text' events from the model

    r2 = s.turn("Now look at the auth flow")
    print(r2.duration_ms)     # 12345
    for ev in r2.events:
        print(ev["type"], ev.get("content"))

# Session is closed automatically here, even if an exception was raised
# inside the block.

Manual form (explicit lifecycle)

For callers that need a Session to outlive a single block (background workers, cross-handler state):

s = forge.create_session(agent="claude", meta={"caller": "petalparse"})
try:
    r = s.turn("hello")
finally:
    s.close()  # idempotent — safe to call repeatedly, including on a
               # session that was already closed by the server's TTL sweeper

Session.close() is idempotent: subsequent calls short-circuit without an HTTP request, and the server itself returns 200 (already_closed: true) on a repeat DELETE. Safe to drop in finally and __exit__ blocks without try/except.

TurnResult shape

@dataclass(frozen=True)
class TurnResult:
    ok: bool
    session_id: str
    turn_index: int                    # 1-based, monotonic per session
    events: list[dict]                 # 'text' | 'tool_call' | 'thinking' | ...
    stop_reason: str | None            # 'end_turn' | 'timeout' | 'error' | ...
    duration_ms: int

    def text(self) -> str: ...         # concat of all 'text' events' content

events is passed through verbatim from the v0.2 server — see the server doc for the exhaustive event-type schema. text() is sugar for "give me the model's prose reply, ignore thinking and tool calls."

Listing & inspection

# All sessions for the calling token (newest first):
sessions = forge.list_sessions()
for row in sessions:
    print(row["session_id"], row["turn_count"], row["closed_at"])

# Drop closed ones:
live = forge.list_sessions(include_closed=False)

# Fresh state for a specific session:
state = forge.get_session("sess_abc")
# {'session_id': 'sess_abc', 'agent': 'claude', 'cwd': ...,
#  'created_at': ..., 'last_turn_at': ..., 'turn_count': 3,
#  'closed_at': None, 'live': True, 'meta': {...}}

# Same thing from inside a block:
with forge.session() as s:
    s.turn("hi")
    print(s.state()["turn_count"])

Errors

The same exception classes apply as v0.1. Two cases worth calling out:

• Cross-token access → ForgeAPIError(404). If token A asks for token B's session, the server returns 404 (not 403) so it can't be used to enumerate live session ids across token boundaries.
• Session closed mid-flight → ForgeAPIError(410). The TTL sweeper or another caller's explicit close() may have ended the session between turns.

from clawdforge import ForgeAPIError

try:
    r = s.turn("hello")
except ForgeAPIError as e:
    if e.status_code == 410:
        # Session expired or was closed — start a fresh one.
        ...
    elif e.status_code == 404:
        # Server forgot us, or we're asking for someone else's id.
        ...
    raise

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.

session(*, agent="claude", meta=None) -> ContextManager[Session] (v0.2)

Block-form session — auto-closes on __exit__. See the "Multi-turn / Sessions (v0.2)" section above.

create_session(*, agent="claude", meta=None) -> Session (v0.2)

Manual-lifecycle session. Caller is responsible for s.close() (typically in finally). close() is idempotent.

list_sessions(*, include_closed=True) -> list[dict] (v0.2)

List sessions for the calling token. Pass include_closed=False to skip closed rows.

get_session(session_id) -> dict (v0.2)

Fetch fresh server-side state for a session. Raises ForgeAPIError(404) if the id is unknown or belongs to another token.

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".