Python SDK

The aiworkforce package builds on the official openai client and adds workflow tracing, business outcomes, human approvals and operator controls. It needs Python 3.10 or later, openai>=1.40 and httpx>=0.27, is fully typed, and has no side effects on import. The optional anthropic extra adds the official Anthropic SDK (anthropic>=0.40) for aiw.anthropic().

Configure#

export AIW_API_KEY="aiw_sk_live_…"export AIW_BASE_URL="https://gateway.worksi.ai"      # the gateway root; the SDK appends /v1

from aiworkforce import AIW aiw = AIW(service="billing-agent")   # service = the default agent nameclient = aiw.openai()                # an openai.OpenAI bound to the gatewayaclient = aiw.async_openai()         # the AsyncOpenAI equivalent

Create one AIW per process and reuse it. It is thread-safe and asyncio-safe, and it flushes pending telemetry at interpreter exit.

Trace a workflow#

from aiworkforce import AIW # reads AIW_API_KEY and AIW_BASE_URLaiw = AIW(service="support-bot")# the OpenAI client, through the gatewayclient = aiw.openai() with aiw.run("customer-chat",             agent="support-bot") as run:    with aiw.step("classify"):        client.chat.completions.create(            model="gpt-5-mini",            messages=history,        )    # tool_call span with aiw.args_hash    with aiw.tool(        "lookup_order",        args={"order_id": order_id},        data_source="orders-db",    ) as t:        t.set_result(orders_db[order_id])    # raises if the agent was paused    aiw.check_status()    run.outcome("ticket_resolved")

• aiw.run(workflow, agent=…) emits run.started when the run begins and a workflow_run span when it ends. The context manager ends the run on errors too; stuck-workflow detection depends on it.
• Model calls made through aiw.openai() or aiw.anthropic() inside a step become children of that step: the gateway creates the llm_call span with your run, step and agent attached.
• aiw.tool(name, args=…) records a tool_call span with a hash of the arguments, which repeated-tool-call detection and tool policies use.
• run.outcome(type, success=…) records a business outcome for ROI. Omit value_usd to use the outcome definition’s unit value.
• @aiw.trace_step and @aiw.trace_tool decorate sync and async functions.
• Threads start with an empty context: use aiw.wrap_context(fn) or ContextThreadPoolExecutor. asyncio tasks inherit the context automatically.

Claude and other providers#

The same aiw.openai() client reaches every provider your organization connects: prefix the model with the credential’s route prefix. For code written for the Anthropic SDK, aiw.anthropic() returns an anthropic.Anthropic bound to the gateway’s native Messages API.

# one client, one Worksi key, every providerclient = aiw.openai()client.chat.completions.create(    model="deepseek/deepseek-flash",    messages=history,) # an anthropic.Anthropic bound to the gateway root# (pip install "aiworkforce[anthropic]")claude = aiw.anthropic() with aiw.run("customer-chat", agent="support-bot"):    with aiw.step("answer"):        claude.messages.create(            model="anthropic/claude-sonnet-5",            max_tokens=1024,            messages=history,        )

• aiw.async_anthropic() is the AsyncAnthropic equivalent.
• Requests carry the current run’s attribution headers, like aiw.openai(). No Anthropic key from your environment is sent to the gateway: it calls Anthropic with your organization’s credential.
• Gateway refusals raise the exceptions listed under Errors, chained from the Anthropic SDK’s error. Without the extra, aiw.anthropic() raises ImportError with the install command.
• Direct mode covers OpenAI clients only.

Honor operator controls#

aiw.check_status() asks the gateway whether an operator paused the agent or workflow, or stopped the run, and raises AgentPausedError or RunStoppedError. Calls made through aiw.openai() and aiw.anthropic() raise the same errors when the gateway refuses them. Call check_status() between long steps that don’t call a model.

Human approvals#

from aiworkforce import ApprovalRejectedError try:    aiw.require_approval(        type="tool_call",        title="Refund $250 for order A-1",        risk="high",        tool="issue_refund",        payload={"order_id": "A-1", "amount_usd": 250},        timeout=300,    )except ApprovalRejectedError:    # a person said no: tell the customer, do not refund    raise with aiw.tool(    "issue_refund",    args={"order_id": "A-1", "amount_usd": 250},) as t:    # already approved for this run    t.set_result({"refund_id": "R-1001"})

require_approval() creates the approval, waits for a decision (approvers act in Operations → Approvals) and records an approval.wait span. It fails closed: if the approval cannot be created, it raises. A rejection raises ApprovalRejectedError; a timeout or an expired approval raises ApprovalTimeoutError. An approval for a tool covers that tool for the rest of the run.

Tool policies#

With tool_policies="enforce" (the default), entering aiw.tool() first asks the gateway for the tool’s verdict, cached for 30 seconds:

• Disallowed by an enforced rule: the tool body does not run, a blocked tool_call span is recorded and ToolNotAllowedError is raised.
• Requires approval under an enforced rule: the SDK requests the approval before the tool runs.
• Monitor-mode rules never block or ask. Violations are recorded on the server.

The check fails open: if the gateway cannot answer, the tool runs.

End users#

Pass your end user and their plan in your app when you start a run, and every model call in it carries both:

with aiw.run("customer-chat", agent="support-bot", end_user=user_id_hash, end_user_plan="free"):    ...

See end‑user plans for the policy that enforces per-plan limits.

Direct mode#

When a workload cannot route inference through the gateway, keep calling OpenAI directly and let the SDK record the calls:

import openaifrom aiworkforce import AIW aiw = AIW(service="scoring-agent")# keeps calling OpenAI directly and traces every callclient = aiw.instrument_openai(openai.OpenAI()) with aiw.run("lead-qualification", agent="scoring-agent"):    client.chat.completions.create(        model="gpt-5-mini",        messages=[            {"role": "user", "content": "Score this lead"},        ],    )

Direct-mode calls are traced and priced, but gateway enforcement (allowlists, budgets, rate limits, pause and stop) does not apply to them. Call aiw.check_status() between steps to honor operator controls.

Errors#

Gateway refusals raise typed exceptions, from aiw.openai() and aiw.anthropic() clients alike. All are subclasses of GatewayError with .code, .status_code and .request_id:

Delivery#

Spans and outcomes are exported by a background thread: bounded queues, gzip batches, retries with backoff. The exporter never raises into your code. In short-lived processes call aiw.flush() before exiting; aiw.shutdown() is registered with atexit. aiw.stats() returns the exporter counters.