Python SDK

A thin client for the Forgebench API that handles retries, typed errors and streaming for you.

Your code holds a Forgebench key (sk_...) instead of a provider key, and Forgebench forwards each request to the model you asked for. Every call is budget-checked before money is spent and recorded in the audit log. See Core concepts for terms such as Agent, Budget and Policy.

Install

pip install "git+https://github.com/seedlinglabs/forgebench-sdk.git#subdirectory=sdk-py"

Requires Python 3.9+. The only runtime dependency is httpx.

Connect to the gateway

You never send a provider credential; Forgebench strips one if a client sends it. Pass your API URL explicitly:

from forgebench import Forgebench

client = Forgebench(api_key="sk_...", base_url="https://api.forgebench.ai")

AsyncForgebench is the async client, with the same constructor and resources:

import asyncio
from forgebench import AsyncForgebench

async def main():
    async with AsyncForgebench(api_key="sk_...", base_url="https://api.forgebench.ai") as client:
        resp = await client.chat.completions.create(
            model="mock-gpt",
            messages=[{"role": "user", "content": "hello"}],
        )
        print(resp.choices[0].message.content)

asyncio.run(main())

whoami

Confirms the key is valid and shows the identity it resolves to. Call it at startup to catch a bad key early:

me = client.whoami()
print(me.tenant_id, me.roles, me.scopes)

Chat completions

resp = client.chat.completions.create(
    model="mock-gpt",
    messages=[{"role": "user", "content": "Prove the governed path works."}],
)

print(resp.choices[0].message.content)
print(resp.usage.total_tokens)
print(resp.trace_id)
print(resp.call_id)

trace_id links the response to its audit entry, metering record and trace. call_id identifies this call; see Call lineage. See Make your first call for the full response shape.

Streaming

for chunk in client.chat.completions.create(
    model="mock-gpt",
    messages=[{"role": "user", "content": "Stream me a sentence."}],
    stream=True,
):
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

The budget check runs before the stream opens, so a 402 raises immediately, never mid-stream. A stream is retried only on a 429 or a failure to connect, before it opens.

Call lineage

Pass a response's call_id as parent_call_id on the calls it causes, such as a sub-agent's turn or the next step of a tool loop. Cost rolls up to the root call, and the console shows the run as a tree. It doesn't affect whether a call is allowed or what it costs.

plan = client.chat.completions.create(model="mock-gpt", messages=[...])

step = client.chat.completions.create(
    model="mock-gpt",
    messages=[...],
    parent_call_id=plan.call_id,
)

Tools

Which allowlist authorizes a tool call depends on the tool's name:

• Names starting with mcp_ (such as mcp_time_get_current_time) are checked against the MCP allowlist. Your agent's MCP client calls the server directly.
• Every other name (such as search or send_email) is checked against the Tool Registry. Your own code runs the tool.

Register each tool in the right place, or neither allowlist will authorize it.

Sending a tool schema

create() has no tools or tool_choice parameters; pass them in extra_body. Below, search is a Tool Registry tool and mcp_time_get_current_time is an MCP tool:

resp = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    extra_body={
        "tools": [
            {
                "type": "function",
                "function": {
                    "name": "search",
                    "description": "Search the KB",
                    "parameters": {"type": "object", "properties": {"q": {"type": "string"}}},
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "mcp_time_get_current_time",
                    "description": "Current time in a timezone",
                    "parameters": {"type": "object", "properties": {"tz": {"type": "string"}}},
                },
            },
        ],
        "tool_choice": "auto",
    },
)

for call in resp.choices[0].message.tool_calls or []:
    if call["function"]["name"].startswith("mcp_"):
        result = your_mcp_client.call(call["function"]["name"], call["function"]["arguments"])
    else:
        result = your_tool_dispatch(call["function"]["name"], call["function"]["arguments"])

tool_calls on the response is a list of OpenAI-shaped dicts ({"id", "type", "function": {"name", "arguments"}}). arguments is a JSON string, not parsed.

If the response names a tool this agent isn't bound to, the whole response is refused with 403 tool_not_authorized (or the MCP equivalent); see Errors. Bind the tool on the Tool Registry or MCP page first.

Other fields only reachable through extra_body

These fields also go in extra_body:

Any other field in extra_body is silently dropped.

The agent_tools helper (agent credentials only)

With an agent key, client.agent_tools gives you the agent's current tool allowlist and a dispatch loop. A binding revoked in the console drops out on the next turn, with no redeploy.

• openai_schema() builds the model's tool list from the allowlist (GET /v1/agent-tools).
• dispatch runs each allowed tool call, reports the outcome, and returns the role: "tool" messages to append before the next turn.

tools = client.agent_tools.openai_schema()

turn = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    extra_body={"tools": tools},
)

messages += client.agent_tools.dispatch(
    turn.choices[0].message.tool_calls,
    execute=lambda name, args: your_mcp_client.call(name, args),
    call_id=turn.call_id,
)

dispatch doesn't stop on a failing tool. An exception from execute is reported as that tool's error and returned to the model as {"error": ...}. Forgebench stores only a tool's name and description, so pass argument schemas with openai_schema(parameters={"search": {...json schema...}}).

agent_tools.report(...) is what dispatch calls internally. Use it directly only if you run tools outside an execute() callback, such as an existing async tool runner. Reports (success, error, latency) go to the audit log and traces; the model never sees them.

Agents

Agent runs are governed like any other call: budget-checked at each hop, metered and audited.

agents = client.agents.list()
for a in agents:
    print(a.id, a.name, a.model)

agent = client.agents.create(
    name="Support bot",
    model="mock-gpt",
    system_prompt="You are a helpful support agent.",
)

run = client.agents.run(agent.id, input={"messages": [
    {"role": "user", "content": "Help me reset my password."},
]}, wait=True, timeout=30)
print(run.status, run.output)

Agent → agent (A2A tasks)

An agent bound to another agent can open a task on it. The callee's governed calls are recorded under the task:

task = client.agents.call(
    "qp-research",
    text="10 MCQs on photosynthesis",
    data={"grade": 9},
    parent_call_id=plan.call_id,
)
if task.state == "input_required":
    task = client.agents.call("qp-research", text="CBSE", context_id=task.context_id)
notes = task.artifact_text()

To receive tasks without an inbound port, call client.agents.serve(handle). It blocks and long-polls for this agent's tasks. Pass parent_call_id=task.call_id on calls made inside the handler so they nest under the task:

def handle(task):
    answer = client.chat.completions.create(
        model="mock-gpt",
        messages=[{"role": "user", "content": task.text}],
        parent_call_id=task.call_id,
    )
    return task.done(text=answer.choices[0].message.content)

client.agents.serve(handle)

handle may return a TaskReply (task.done(...), task.ask(...), task.fail(...)), or a str, dict or list of artifacts, which becomes a completed reply. An exception fails the task without stopping the loop.

Runs

run = client.runs.create(model="mock-gpt", input={"prompt": "hello"})
run = client.runs.get(run.id)
run = client.runs.wait(run.id, timeout=30)
print(run.status)

wait polls until the run reaches succeeded or failed.

Workflows

A workflow is a named group of agents (steps) that one app runs in order, one agent and one key per step. Register it once, then drive it with client.workflow:

me = client.whoami()
created = client.workflows.register(
    name="Question paper",
    owner_identity_id=me.identity_id,
    steps=[{"name": "grading", "model": "mock-gpt"}, {"name": "retry", "model": "mock-gpt"}],
    edges=[{"from": "grading", "to": "retry", "kind": "sequence"}],
)
keys = {c["step"]: c["credential"]["api_key"] for c in created["credentials"]}  # shown once

wf = client.workflow("question-paper", keys=keys)
with wf.run() as run:
    run.step("grading").chat(model="mock-gpt", messages=[...])   # uses grading's key
    run.step("retry").chat(model="mock-gpt", messages=[...])     # parent = grading's call
    print(run.root_call_id)

client.workflows.checks("question-paper")   # wired? threaded? seeing traffic?

Each step call uses that step's own key and sends the previous call of the run as its parent (X-Parent-Call); the first call is a root. AsyncForgebench has the same surface (async with wf.run() as run). client.workflows also has list, get, update, archive, add_step, remove_step, add_edge and remove_edge.

Tracing

Mark your agent's steps and each run becomes one nested trace: the agent, its functions and tools, and each model call with its tokens and cost. Steps are sent in the background with the same API key. Read traces in the console. See Tracing your agent.

Mark a function

@observe records each call of a function (sync or async): arguments as input, return value as output, duration, and any exception as an error (re-raised unchanged). Model calls inside it nest under it.

from forgebench import Forgebench, observe

fb = Forgebench()

@observe(as_type="tool")
def lookup_order(order_id: str) -> dict:
    return orders.get(order_id)

@observe(as_type="agent", name="support-agent")
def run(question: str) -> str:
    fb.update_current_trace(user_id="customer-42", session_id="chat-7", tags=["support"])
    order = lookup_order("A-1001")
    reply = fb.chat.completions.create(model="mock-gpt", messages=[...])  # nested
    return reply.choices[0].message.content

name defaults to the function's name; as_type is one of span (default), agent, tool, chain, retriever, embedding, evaluator, guardrail, event. Set capture_input=False or capture_output=False for values that are large or must not leave the process. Use fb.observe(...) to bind the decorator to one client when your app uses several API keys.

Mark a block of code

with fb.start_as_current_observation("plan", as_type="chain", input={"q": question}) as plan:
    steps = make_plan(question)
    plan.update(output={"steps": steps}, metadata={"planner": "v2"})

Steps and model calls made inside the block nest under it. An exception is recorded on the step and raised again.

Steps that start and end in different places

fb.start_observation(...) creates a step that isn't the current one, for a callback or background job. step.start_observation(...) adds a child. Call end() when done; a step that never ends is never sent.

job = fb.start_observation("nightly-sync", as_type="chain")
page = job.start_observation("fetch-page", as_type="retriever")
page.update(output={"rows": 120})
page.end()
job.end()

Threads and asyncio

asyncio tasks keep the current step on their own. Work handed to a thread pool needs bind_context:

from concurrent.futures import ThreadPoolExecutor
from forgebench import bind_context

with ThreadPoolExecutor() as pool:
    results = list(pool.map(bind_context(lookup_order), order_ids))

Configuration

from forgebench import Forgebench, TracingConfig

fb = Forgebench(tracing=TracingConfig(
    sample_rate=0.2,             # keep 20% of traces (whole traces, never half)
    environment="production",
    release="2.3.0",
    mask=my_mask,                # runs on every value and error message before it leaves
))

Without tracing=, the client reads FORGEBENCH_TRACING_ENABLED, FORGEBENCH_SAMPLE_RATE, FORGEBENCH_ENVIRONMENT and FORGEBENCH_RELEASE; a value that does not parse is ignored with a warning. enabled=False turns tracing into a no-op without touching your code.

Sending before exit

Steps are sent about every second. In a short script, call fb.flush() before exit; fb.close() or leaving a with Forgebench() as fb: block also sends what's left. A failed send is logged and dropped, never raised.

Voice

Transcription, Speech Synthesis and streaming agent runs are governed like chat. See Govern your voice agent.

try:
    heard = client.audio.transcriptions.create(
        open("caller.wav", "rb"), "gpt-4o-transcribe", session="call-8841"
    )                                   # .text, .raw, .trace_id
    spoken = client.audio.speech.create(
        "gpt-4o-mini-tts", "Hello", "alloy", session="call-8841"
    )                                   # .audio (bytes), .content_type, .trace_id
    for event in client.agents.run(agent_id, message="Hi", stream=True):
        print(event)                    # chat.completion.chunk dicts, then {"type": "agent.run.completed", "answer", "trace_id", "cost_usd", "citations"}
except BudgetExceededError:
    print("over budget")

file accepts bytes, a path, a binary file object or a (filename, bytes) tuple. session sets X-Forgebench-Session. AsyncForgebench has the same methods; await client.agents.run(..., stream=True) returns an async iterator. Admin helpers: client.voice.policy.get() / update(...), client.voice.assets.list() / create(...) / update(...) / revoke(id) and client.voice.consents.list() / create(...) / revoke(id). See Voice governance.

A streaming agents.run raises StreamError (e.code, e.message) if the server sends an event: error frame mid-stream, or if the stream ends without agent.run.completed / [DONE] (code="stream_truncated"). To clear a Voice Asset field pass it as None, for example assets.update(id, expires_at=None). voice_headers(...) builds the session, turn and traceparent headers for your own pipeline's calls; AUDIO_TRACE_ID_HEADER is the /v1/audio/* trace-id response header (chat uses TRACE_ID_HEADER).

Errors

Every SDK-raised error is a forgebench.ForgebenchError. Network failures raise ForgebenchConnectionError directly; anything that got an HTTP response raises a subclass of APIError, one per status code, each carrying .status_code, .code, .message, .body and .request_id:

By the time you see a RateLimitError, the SDK has already retried it up to max_retries times (default 2). The same applies to a ServerError or connection failure on a read.

A write (chat call, A2A task, tool report) is never replayed after a 5xx or a dropped connection, because the server may already have run and billed it. A ServerError from a chat call means it may or may not have run; check before repeating it.

Other errors will fail again if retried unchanged. Three unauthorized attempts at the same tool within ten minutes escalate to a 429 (tool_repeatedly_unauthorized).

PermissionDeniedError reports its cause in different places. Tool refusals set .code (for example "tool_not_authorized"). For model_not_allowed, .code is None and the identifier is in .body["detail"]["error"]:

from forgebench import (
    Forgebench,
    AuthenticationError,
    BudgetExceededError,
    PermissionDeniedError,
    NotFoundError,
    ConflictError,
    ValidationError,
    RateLimitError,
    ServerError,
    ForgebenchConnectionError,
)

try:
    resp = client.chat.completions.create(model="gpt-4o", messages=msgs, extra_body={"tools": tools})
except BudgetExceededError as e:
    log.warning("budget gate refused the call: %s", e)
except PermissionDeniedError as e:
    log.error("not authorized: %s", e.body)
except AuthenticationError as e:
    log.error("auth failed: %s", e)
except ValidationError as e:
    log.error("request rejected: %s", e.body)
except NotFoundError as e:
    log.error("not found: %s", e)
except ConflictError as e:
    log.error("conflicting state: %s", e)
except RateLimitError as e:
    log.warning("rate limited after retries: %s", e)
except ServerError as e:
    log.error("control-plane error after retries: %s", e)
except ForgebenchConnectionError as e:
    log.error("could not reach the control plane: %s", e)

Every 402, 403, 400 and 429 refusal is written to the Refusals log with its cause. See Errors and refusals for each body shape.

Retries

The Python, TypeScript and Java SDKs share one retry rule: a call the server may already have run and billed is never sent twice.

• A stream that is already delivering events is never retried.
• A read timeout on a write is not retried: the server may have run the call.
• There are at most max_retries retries (default 2). The wait starts at 0.5s and doubles up to 8s, with jitter. A Retry-After header in seconds replaces it, capped at 30s; an HTTP-date is ignored.

Metering and budget state

s = client.metering_summary()
print(f"spent ${s.spent_usd} / limit ${s.monthly_limit_usd} "
      f"(remaining ${s.remaining_usd})")
print(s.by_model)

Next