Python SDK

Full reference for the Thoth Python SDK — ThothConfig, instrument(), instrument_toolchain(), toolchain_function_map(), instrument_anthropic(), instrument_openai(), instrument_claude_agent_sdk(), LangGraph and CrewAI integrations.

Installation

pip install aten-thoth

Python 3.12+ required. The package ships type hints and is fully typed.

API Style

Preferred APIs for new integrations:

thoth.instrument()
thoth.instrument_toolchain()
thoth.toolchain_function_map()
thoth.instrument_anthropic()
thoth.instrument_openai()
thoth.instrument_claude_agent_sdk()

Legacy compatibility:

from thoth import ThothClient remains supported for backward compatibility.
ThothClient.wrap(...) is a legacy alias to ThothClient.instrument(...).

from thoth import ThothClient
 
client = ThothClient(
    agent_id="support-bot",
    approved_scope=["search_docs"],
    tenant_id="acme-corp",
    api_url="https://enforce.acme.example",
)
 
agent = client.wrap(agent)  # legacy alias

`thoth.instrument()`

Instrument a generic AI agent with Thoth governance. Wraps all tools with enforce/emit hooks. Returns the same agent object (mutated in-place).

def instrument(
    agent: Any,
    *,
    agent_id: str,
    approved_scope: list[str],
    tenant_id: str,
    user_id: str = "system",
    enforcement: str = "block",
    api_key: str | None = None,
    api_url: str | None = None,
    session_id: str | None = None,
    session_intent: str | None = None,
    environment: str = "prod",
    enforcement_trace_id: str | None = None,
) -> Any

Parameters

Parameter	Type	Required	Description
`agent`	`Any`	Yes	Agent object with a `.tools` attribute
`agent_id`	`str`	Yes	Unique agent identifier
`approved_scope`	`list[str]`	Yes	Authorization allow-list of tool names the agent is permitted to execute
`tenant_id`	`str`	Yes	Your Thoth tenant identifier
`user_id`	`str`	No	User initiating the session (default: `"system"`)
`enforcement`	`str`	No	Enforcement mode (default: `"block"`)
`api_key`	`str \| None`	No	API key; falls back to `THOTH_API_KEY` env var
`api_url`	`str \| None`	Yes*	Tenant API base URL used for both event ingestion and policy checks; provide directly or via `THOTH_API_URL`
`session_id`	`str \| None`	No	Custom session ID; auto-generated if omitted
`session_intent`	`str \| None`	No	Session purpose for HIPAA minimum-necessary scope checks
`environment`	`str`	No	Environment tag for env-scoped policy lookup (default: `"prod"`)
`enforcement_trace_id`	`str \| None`	No	Optional cross-service correlation ID; defaults to session ID when omitted

* api_url may be omitted only when THOTH_API_URL is set.

approved_scope clarifier:

It is an authorization allow-list, not a free-text description field.
Values must match emitted tool names exactly (for example search_docs or data_sources.cloudtrail).
Use tool schemas/prompts and policy context fields (session_intent, purpose, data_classification, task_context) to convey intent/context to the enforcer.

Supported agent shapes

instrument() detects the agent type automatically:

LangChain AgentExecutor — calls wrap_langchain_agent() internally
CrewAI Agent — calls wrap_crewai_agent() internally
Generic — any object with a .tools list where each tool has a .name and .run() method

# LangGraph / LangChain
from langchain.agents import AgentExecutor
instrument(agent_executor, agent_id="...", approved_scope=["..."], tenant_id="...")
 
# CrewAI
from crewai import Agent
instrument(crew_agent, agent_id="...", approved_scope=["..."], tenant_id="...")
 
# Generic (LlamaIndex, custom, etc.)
agent.tools = [MyTool("search"), MyTool("compute")]
instrument(agent, agent_id="...", approved_scope=["search"], tenant_id="...")

Enforcement Modes

👁 observe

Observe

Log all tool calls. Never block or interrupt. Use for baselining agent behavior.

🔐 step_up

Step-Up Auth

Require human approval for out-of-scope calls before execution proceeds.

📈 progressive

Progressive

Escalate automatically: warn → step-up → block after repeated violations.

🛑 block

Block

Immediately reject any tool call outside the approved scope. No exceptions.

`instrument_anthropic()`

Wrap tool functions for use in an Anthropic Claude agentic loop.

def instrument_anthropic(
    tool_fns: dict[str, Callable],
    *,
    agent_id: str,
    approved_scope: list[str],
    tenant_id: str,
    user_id: str = "system",
    enforcement: str = "block",
    api_key: str | None = None,
    api_url: str | None = None,
    session_id: str | None = None,
    session_intent: str | None = None,
    environment: str = "prod",
    enforcement_trace_id: str | None = None,
) -> dict[str, Callable]

Returns a new dict[str, Callable] with governance-wrapped callables. The wrapped callables receive the block.input dict from Anthropic tool-use responses.

import anthropic
from thoth import instrument_anthropic
 
def search_docs(input: dict) -> str:
    return f"Results for: {input['query']}"
 
governed = instrument_anthropic(
    {"search_docs": search_docs},
    agent_id="support-bot",
    approved_scope=["search_docs"],
    tenant_id="acme-corp",
)
 
client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=[{"name": "search_docs", "description": "Search company docs",
            "input_schema": {"type": "object", "properties": {"query": {"type": "string"}}}}],
    messages=[{"role": "user", "content": "How do I reset my password?"}],
)
for block in response.content:
    if block.type == "tool_use":
        fn = governed.get(block.name)
        if fn:
            result = fn(block.input)  # governance enforced here

`instrument_toolchain()`

Recursively wraps nested dict/list/object toolchains from a single call. Public methods are named by dotted path (for example data_sources.cloudtrail) and governed through the same enforcement path as other SDK integrations. max_depth is optional; when omitted (None), traversal automatically covers the full reachable toolchain graph (cycles are skipped).

def instrument_toolchain(
    toolchain: Any,
    *,
    agent_id: str,
    approved_scope: list[str],
    tenant_id: str,
    user_id: str = "system",
    enforcement: str = "block",
    api_key: str | None = None,
    api_url: str | None = None,
    session_id: str | None = None,
    session_intent: str | None = None,
    environment: str = "prod",
    enforcement_trace_id: str | None = None,
    include_private: bool = False,
    max_depth: int | None = None,
) -> Any

import thoth
 
class DataSources:
    def cloudtrail(self, args: dict) -> list[dict]:
        return self.parse(args)
 
    def parse(self, args: dict) -> list[dict]:
        return [{"signal_id": args["signal_id"]}]
 
toolchain = {"data_sources": DataSources()}
 
governed = thoth.instrument_toolchain(
    toolchain,
    agent_id="alert-triage-agent",
    approved_scope=["data_sources.cloudtrail", "data_sources.parse"],
    tenant_id="acme-corp",
    enforcement="observe",
)
 
result = governed["data_sources"].cloudtrail({"signal_id": "sig-1"})

`toolchain_function_map()`

Builds a framework-compatible function map from a governed toolchain so users do not need to hand-write mapping dictionaries.

def toolchain_function_map(
    toolchain: Any,
    *,
    include_private: bool = False,
    max_depth: int | None = None,
) -> dict[str, Callable]

import thoth
 
governed = thoth.instrument_toolchain(...)
function_map = thoth.toolchain_function_map(governed)

This is useful for frameworks like AutoGen that require a function_map boundary even when your tools are already instrumented.

`instrument_openai()`

Wrap tool functions for use in an OpenAI tool-calling loop. Signature is identical to instrument_anthropic() (including session_intent, environment, and enforcement_trace_id).

import json
import openai
from thoth import instrument_openai
 
def search_docs(args: dict) -> str:
    return f"Results for: {args['query']}"
 
governed = instrument_openai(
    {"search_docs": search_docs},
    agent_id="support-bot",
    approved_scope=["search_docs"],
    tenant_id="acme-corp",
)
 
client = openai.OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    tools=[{"type": "function", "function": {"name": "search_docs",
            "parameters": {"type": "object", "properties": {"query": {"type": "string"}}}}}],
    messages=[{"role": "user", "content": "How do I reset my password?"}],
)
for tool_call in (response.choices[0].message.tool_calls or []):
    fn = governed.get(tool_call.function.name)
    if fn:
        args = json.loads(tool_call.function.arguments)
        result = fn(args)  # governance enforced here

`instrument_claude_agent_sdk()`

Instrument claude-agent-sdk by attaching Thoth governance to ClaudeAgentOptions.can_use_tool plus post-success/post-failure tool hooks.

def instrument_claude_agent_sdk(
    options: Any | None = None,
    *,
    agent_id: str,
    approved_scope: list[str],
    tenant_id: str,
    user_id: str = "system",
    enforcement: str = "block",
    api_key: str | None = None,
    api_url: str | None = None,
    session_id: str | None = None,
    session_intent: str | None = None,
    environment: str = "prod",
    enforcement_trace_id: str | None = None,
    emit_tool_lifecycle_hooks: bool = True,
) -> Any

Notes:

claude-agent-sdk requires streaming mode for can_use_tool, so pass an async iterable prompt to query(...) instead of a plain string prompt.
This integration is for query() / ClaudeSDKClient flows that do not expose a .tools list.

import anyio
from claude_agent_sdk import ClaudeAgentOptions, query
import thoth
 
async def prompt_stream():
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Read README.md and summarize risks."},
        "parent_tool_use_id": None,
        "session_id": "demo-session",
    }
 
async def main():
    options = ClaudeAgentOptions(max_turns=1, allowed_tools=["Read"])
    options = thoth.instrument_claude_agent_sdk(
        options,
        agent_id="claude-agent-sdk-demo",
        approved_scope=["Read"],
        tenant_id="acme-corp",
    )
    async for message in query(prompt=prompt_stream(), options=options):
        print(message)
 
anyio.run(main)

LangGraph Integration

import os
from thoth.models import ThothConfig
from thoth.session import SessionContext
from thoth.emitter import HttpEmitter
from thoth.enforcer_client import EnforcerClient
from thoth.step_up import StepUpClient
from thoth.tracer import Tracer
 
config = ThothConfig(
    agent_id="langgraph-agent",
    approved_scope=["search", "retrieve", "summarize"],
    tenant_id="acme-corp",
    api_url=os.environ["THOTH_API_URL"],
)
session = SessionContext(config)
tracer = Tracer(
    config=config,
    session=session,
    emitter=HttpEmitter(api_url=config.resolved_api_url, api_key=config.api_key or ""),
    enforcer=EnforcerClient(config),
    step_up=StepUpClient(config),
)
 
# Wrap LangGraph tool callables directly
governed_search = tracer.wrap_tool("search", search)
governed_retrieve = tracer.wrap_tool("retrieve", retrieve)

CrewAI Integration

from crewai import Agent
from thoth import instrument
 
researcher = Agent(
    role="Senior Research Analyst",
    goal="Research quarterly earnings",
    tools=[search_tool, scrape_tool],
)
 
instrument(
    researcher,
    agent_id="crewai-researcher",
    approved_scope=["search_tool", "scrape_tool"],
    tenant_id="acme-corp",
    enforcement="block",
)

Error Handling

`ThothPolicyViolation`

Raised when the enforcer blocks a tool call.

from thoth import ThothPolicyViolation
 
try:
    result = tool_fn(args)
except ThothPolicyViolation as e:
    print(e.tool_name)     # "submit_payment"
    print(e.reason)        # "tool not in approved scope"
    print(e.violation_id)  # "vio_abc123" — correlates with Thoth audit records
    print(e.decision_reason_code)   # stable machine-readable reason
    print(e.action_classification)  # read/write/execute class
    print(e.risk_score)             # risk score from decisioning
    print(e.pack_id, e.pack_version, e.rule_version)
    print(e.matched_rule_ids, e.matched_control_ids)
    print(e.policy_references, e.model_signals)

Current SDK behavior is fail-closed for enforcement: if the enforcer is unreachable, tool calls fall back to BLOCK (reason="enforcer unavailable").

`EnforcementDecision`

Normalized enforcer response model (snake_case and camelCase payload aliases are both accepted):

from thoth.models import EnforcementDecision, DecisionType
 
decision = EnforcementDecision.model_validate(payload)
if decision.decision == DecisionType.MODIFY:
    print(decision.modified_tool_args, decision.modification_reason)
elif decision.decision == DecisionType.DEFER:
    print(decision.defer_reason, decision.defer_timeout_seconds)

Important fields include:

decision, authorization_decision, reason, violation_id, hold_token
decision_reason_code, action_classification
risk_score, latency_ms
pack_id, pack_version, rule_version
regulatory_regimes, matched_rule_ids, matched_control_ids
policy_references, model_signals, receipt

Environment Variables

Variable	Description
`THOTH_API_KEY`	API key for hosted Thoth authentication
`THOTH_API_URL`	Required tenant API base URL used for both enforcement and event ingestion
`THOTH_LOG_LEVEL`	Optional SDK logger level override (`DEBUG`, `INFO`, `WARNING`, `ERROR`). Falls back to `LOG_LEVEL` when unset.

If THOTH_API_URL is set, it overrides the api_url value passed in config.

Python SDK

On this page