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.

Reference examples

Start with Your first governed agent for a complete runnable example and tested package version. Examples here focus on individual APIs and may depend on callbacks or clients defined by your application. Replace illustrative tenant and endpoint values with the supplied connection details; agree on server policy before using real tools.

Installation

pip install atensec-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 a list of intended tool names, not a policy installation or authorization grant.
Values must match emitted tool names exactly (for example search_docs or data_sources.cloudtrail).
In the hosted trial, the server policy defines allowed tools. Outside that list, execution requires human approval even if the model reports low risk. Explicit blocks and stronger guards still deny execution. Observe mode monitors only. This scope floor applies to a nonempty allowed-tool list; an empty list leaves the other policy and risk checks in control.
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

# LangChain (LangGraph uses instrument_langgraph)
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

Collect telemetry without enforcing actions. Python bypasses the enforcer in this mode.

step_up

Step-up auth

Request approval according to server policy. Keep the tool paused until the hold resolves.

progressive

Progressive

Evaluate actions with progressive policy. Outcomes depend on server policy and context.

block

Block

Use blocking enforcement. Server policy decides which actions are permitted or refused.

`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="block",
)
 
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

from thoth import instrument_langgraph
 
governed_tools = instrument_langgraph(
    [search, retrieve],
    agent_id="langgraph-agent",
    approved_scope=["search", "retrieve"],
    tenant_id="acme-corp",
    enforcement="step_up",
    api_url="https://enforce.acme.example",
)

See the full guide: /docs/sdks/langgraph-integration

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)

With the default THOTH_FAIL_OPEN=false, enforcing modes fail closed: 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