Python SDK

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

Installation

pip install aten-thoth

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

`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 = "progressive",
    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	Authorized tool names
`tenant_id`	`str`	Yes	Your Thoth tenant identifier
`user_id`	`str`	No	User initiating the session (default: `"system"`)
`enforcement`	`str`	No	Enforcement mode (default: `"progressive"`)
`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.

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 = "progressive",
    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_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

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

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

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

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

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

Python SDK

On this page