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:
1. **LangChain `AgentExecutor`** — calls `wrap_langchain_agent()` internally
2. **CrewAI `Agent`** — calls `wrap_crewai_agent()` internally
3. **Generic** — any object with a `.tools` list where each tool has a `.name` and `.run()` method
```python
# 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.
```python
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.
```python
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).
```python
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
```
```python
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.
```python
def toolchain_function_map(
toolchain: Any,
*,
include_private: bool = False,
max_depth: int | None = None,
) -> dict[str, Callable]
```
```python
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`).
```python
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.
```python
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.
```python
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
```python
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
```python
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.
```python
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):
```python
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