Framework Integrations

Three ways to connect the SDK to your LLM stack. The core SDK is framework-agnostic; adapters add convenience for popular frameworks.

OpenAI SDK

The wrap_openai function patches an OpenAI client to automatically log audit events, track cost, and enforce budget caps on every chat.completions.create() call. Works with both sync and async clients. AsyncOpenAI is detected via iscoroutinefunction on the create method or "Async" in the client class name.

Installbash

pip install "code-atelier-governance[openai]"

Usagepython

from openai import AsyncOpenAI
from codeatelier_governance import GovernanceSDK
from codeatelier_governance.integrations.openai_wrap import wrap_openai

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    client = AsyncOpenAI()
    wrap_openai(client, sdk=sdk, agent_id="my-agent")

    # Every call now auto-logs audit events + tracks cost:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello"}],
    )
    # audit rows: llm.call → llm.result (with model + token count)
    # cost.track() called automatically with actual token usage

What auto-populates: model name, token counts (prompt + completion), cost in USD (via built-in pricing table). What you control: agent_id, session_id (via context manager), whether to run scope.check() and cost.check_or_raise() before the call.

Auto USD via built-in pricingv0.2.2 — The wrapper now auto-computes USD from the model name using the built-in pricing table (24 models). No need to pass usd= manually.

Double-wrap protection — Calling wrap_openai() twice on the same client is now a no-op. The second call logs a warning and returns the already-wrapped client unchanged. No duplicate audit events, no double cost tracking.

Streaming responsesv0.5.0 — When stream=True is passed, a warning is logged and cost tracking is bypassed. Token usage is not available until the stream is fully consumed. Call sdk.cost.track() manually after consuming the stream.

Anthropic SDK

v0.4.0

The wrap_anthropic function patches an Anthropic client to automatically log audit events, track cost, and enforce budget caps on every messages.create() call. Works with both sync (anthropic.Anthropic) and async (anthropic.AsyncAnthropic) clients.

Installbash

pip install "code-atelier-governance[anthropic]"

Usagepython

import anthropic
from codeatelier_governance import GovernanceSDK
from codeatelier_governance.integrations.anthropic_wrap import wrap_anthropic

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    client = anthropic.AsyncAnthropic()
    wrap_anthropic(client, sdk=sdk, agent_id="my-agent")

    # Every call now auto-logs audit events + tracks cost:
    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}],
    )
    # audit rows: llm.call -> llm.result (with model + token count)
    # cost.track() called automatically with actual token usage

What auto-populates: model name, token counts (input + output), USD cost estimation (via built-in pricing table), budget enforcement (check_or_raise before each call). What you control: agent_id, session_id, model selection.

Auto USD via built-in pricing - The wrapper auto-computes USD from the model name using the built-in pricing table. No need to pass usd= manually. Input and output tokens are extracted from the Anthropic response and costed separately.

Double-wrap protection - Calling wrap_anthropic() twice on the same client is a safe no-op. The second call logs a warning and returns the already-wrapped client unchanged. No duplicate audit events, no double cost tracking.

Streaming responsesv0.5.0 — When stream=True is passed or a streaming response object is detected (MessageStream, AsyncMessageStream), a warning is logged and cost tracking is bypassed. Call sdk.cost.track() manually after consuming the stream.

LangChain

The GovernanceCallbackHandlermaps LangChain's callback hooks to governance audit events. Every LLM call, tool call, chain, and agent action becomes an audit row automatically.

Installbash

pip install "code-atelier-governance[langchain]"

Observation mode (default)python

from langchain.chat_models import ChatOpenAI
from codeatelier_governance import GovernanceSDK
from codeatelier_governance.integrations.langchain_handler import GovernanceCallbackHandler

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    handler = GovernanceCallbackHandler(sdk=sdk, agent_id="langchain-agent")
    llm = ChatOpenAI(callbacks=[handler])

    # Every LLM/tool/chain event maps to an audit row:
    # on_llm_start    → audit(kind="llm.call")
    # on_llm_end      → audit(kind="llm.result") + cost.track()
    # on_tool_start   → scope.check() + audit(kind="tool.call")
    # on_tool_end     → audit(kind="tool.result")
    # on_chain_start  → audit(kind="chain.start")
    # on_agent_action → audit(kind="agent.action")

Enforcement mode

v0.2.1

Pass enforce=True to make the handler raise ScopeViolation through LangChain's callback system when a tool call fails scope checks. This is the recommended setting for production — observation mode logs violations but cannot prevent them.

Enforcement mode (recommended for production)python

from codeatelier_governance.integrations.langchain_handler import GovernanceCallbackHandler
from codeatelier_governance.scope import ScopePolicy

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    # Register the agent's allowed tools
    sdk.scope.register(ScopePolicy(
        agent_id="langchain-agent",
        allowed_tools=frozenset({"search", "calculator"}),
    ))

    # enforce=True makes scope violations raise through the callback
    handler = GovernanceCallbackHandler(
        sdk=sdk,
        agent_id="langchain-agent",
        enforce=True,
    )
    llm = ChatOpenAI(callbacks=[handler])

    # Tool calls not in the whitelist now raise ScopeViolation
    # instead of just logging the violation

Hidden tool filteringv0.2.2 — In enforcement mode, the handler also filters hidden tools from on_llm_start invocation params. Tools listed in hidden_toolsare removed from the LLM's tool list before the call fires, so the agent never even knows they exist.

Raw Python (no framework)

Don't use LangChain, OpenAI, or Anthropic? The core SDK works with any Python code via decorators and context managers. No adapter needed.

Decorator patternpython

@sdk.audit.track(kind="process_document", agent_id="doc-agent")
async def process_document(doc_id: str) -> dict:
    # Audit: process_document.start logged with input_hash
    result = await my_llm_call(doc_id)
    # Audit: process_document.end logged with output_hash
    return result

Manual instrumentationpython

from codeatelier_governance.audit import AuditEvent

with sdk.audit.session() as session_id:
    # Filter hidden tools before passing to LLM:
    visible_tools = sdk.scope.filter_tools("my-agent", all_tools)

    await sdk.scope.check(agent_id="my-agent", tool="send_email")
    await sdk.cost.check_or_raise("my-agent", session_id)

    await sdk.audit.log(AuditEvent(
        agent_id="my-agent",
        kind="email.send",
        model="gpt-4o-mini",
        metadata={"recipient": "user@example.com"},
    ))

    # ... your business logic ...

    # Auto-compute USD from model name:
    await sdk.cost.track_usage("my-agent", session_id,
        model="gpt-4o-mini", input_tokens=30, output_tokens=20)

Multi-Agent Setup

v0.5.0

In a multi-agent deployment, use a separate wrapped client per agent. Each agent gets its own scope, budget, and audit trail. The SDK instance is shared — only the agent_id differs.

Multi-agent examplepython

from openai import AsyncOpenAI
from codeatelier_governance import GovernanceSDK, ScopePolicy, BudgetPolicy
from codeatelier_governance.integrations.openai_wrap import wrap_openai

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    # Agent 1: billing (restricted scope, low budget)
    sdk.scope.register(ScopePolicy(
        agent_id="billing-agent",
        allowed_tools=frozenset({"read_invoice", "send_email"}),
    ))
    sdk.cost.register(BudgetPolicy(
        agent_id="billing-agent", per_session_usd=1.00,
    ))
    billing_client = wrap_openai(AsyncOpenAI(), sdk=sdk, agent_id="billing-agent")

    # Agent 2: research (wider scope, higher budget)
    sdk.scope.register(ScopePolicy(
        agent_id="research-agent",
        allowed_tools=frozenset({"search", "summarize", "web_browse"}),
    ))
    sdk.cost.register(BudgetPolicy(
        agent_id="research-agent", per_session_usd=5.00,
    ))
    research_client = wrap_openai(AsyncOpenAI(), sdk=sdk, agent_id="research-agent")

OpenTelemetry Export

Feed audit events into your existing OTel stack (Datadog, Honeycomb, Grafana Tempo) via the optional OTel exporter. Each audit event becomes an OTel span with GenAI semantic convention attributes.

Install + Usagepython

pip install "code-atelier-governance[otel]"

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from codeatelier_governance.audit.otel_exporter import OTelExporter

trace.set_tracer_provider(TracerProvider())  # configure your exporter
exporter = OTelExporter()
sdk.audit.subscribe(exporter)
# Every audit.log() now also emits an OTel span

Continue to Console Guide →