Thoth SDK

Thoth SDK instruments your AI agents for governance, policy enforcement, and behavioral monitoring. Every tool call is evaluated against your organization's security policies in <100ms — blocking, stepping up for human approval, or silently observing based on your configured enforcement mode.

Package name: atensec-thoth | PyPI: pip install atensec-thoth

Table of Contents

1. Installation
2. Quick Start
3. What's New Since 0.1.15
4. Legacy Compatibility (ThothClient)
5. How It Works
6. Integration Examples
   • LangChain AgentExecutor
   • LangGraph
   • Claude Agent SDK (query)
   • OpenAI Function Calling
   • Anthropic Claude
   • CrewAI
   • AutoGen
   • Custom Tools (Generic)
7. Enforcement Modes
8. Policy Decisions
9. Handling Violations
10. Step-Up Authentication
11. Session Inspection
12. Configuration Reference
13. Dashboard

Installation

pip install atensec-thoth

With LangChain / LangGraph support:

pip install "atensec-thoth[langchain]"

With OpenAI support:

pip install "atensec-thoth[openai]"

With Claude Agent SDK support:

pip install "atensec-thoth[claude]"

With AutoGen support:

pip install "atensec-thoth[autogen]"

Requirements: Python 3.12+

Quick Start

1. Get your API key from your Thoth dashboard at https://<tenant>.<apex-domain> under Settings → API Keys.

2. Set environment variables:

export THOTH_API_KEY="thoth_live_..."
export THOTH_API_URL="https://enforce.<tenant>.<apex-domain>"

3. Instrument your agent — three lines of code:

import os
import thoth

# Instrument your agent — returns the same object, mutated in-place
agent = thoth.instrument(
    agent,
    agent_id="document-summarizer",
    approved_scope=["read_file", "summarize"],   # tools outside this list are blocked
    tenant_id=os.environ["THOTH_TENANT_ID"],
    api_url=os.environ["THOTH_API_URL"],
    user_id="alice@example.com",
    enforcement="block",   # default deny
    # api_key set via THOTH_API_KEY env var
)

# Every tool call is now governed — no other changes required
result = agent.run("Summarize the attached document and send it to the team.")

That's it. No AWS credentials, no infrastructure setup — the SDK sends events and enforcement requests over HTTPS to your tenant API URL.

What's New Since 0.1.15

If you're upgrading from 0.1.15, these are the material integration changes:

• Package publish target is now atensec-thoth (docs/examples updated accordingly).
• Decision model expanded beyond ALLOW/BLOCK/STEP_UP:
  • MODIFY: enforcer can sanitize/transform tool args before execution.
  • DEFER: enforcer can defer execution with retry guidance.
• Decision envelope context surfaced end-to-end:
  • decision_reason_code, action_classification, authorization_decision
  • risk_score, latency_ms, pack_id, pack_version, rule_version
  • matched_rule_ids, matched_control_ids, regulatory_regimes, policy_references, model_signals
  • fastml_features, score_components, top_contributors, decision_evidence, receipt
• Async tool wrappers use async enforcement path (acheck) and preserve sync/async behavior.
• Structured tool metadata emission now includes normalized tool_call payload and result summaries.
• Enforcer payload now includes richer context when provided:
  • tool_args, environment, enforcement_trace_id
  • session_intent, purpose, data_classification, task_context
• Release automation moved to tag-driven public workflow in atensecurity/thoth-py, with release notes sourced from CHANGELOG.md.

Legacy Compatibility (ThothClient)

thoth.instrument(), thoth.instrument_toolchain(), thoth.toolchain_function_map(), thoth.instrument_anthropic(), and thoth.instrument_openai() are the preferred APIs for new integrations.

ThothClient is still exported for backward compatibility with older examples and wrappers:

from thoth import ThothClient

client = ThothClient(
    agent_id="document-summarizer",
    approved_scope=["read_file", "summarize"],
    tenant_id="acme-corp",
    api_url="https://enforce.acme.example",
)

agent = client.wrap(agent)  # legacy alias for client.instrument(agent)

For new code, prefer the module-level APIs shown in Quick Start.

How It Works

Agent calls tool
      │
      ▼
 Thoth intercepts (wrap_tool)
      │
      ├── Emits TOOL_CALL_PRE event → Aten API (async, non-blocking)
      │
      ├── Calls enforcer /v1/enforce
      │        │
      │        ├── ALLOW   → tool executes normally
      │        ├── STEP_UP → waits for human approval (polls /v1/enforce/hold/{token})
      │        └── BLOCK   → raises ThothPolicyViolation
      │
      ├── Tool executes (if allowed)
      │
      └── Emits TOOL_CALL_POST event → Aten API (async, non-blocking)

Events are batched and flushed to the Aten ingest API in a background daemon thread with at-most-10 per batch. If the enforcer is unreachable, Thoth fails closed (BLOCK) and logs a warning. Tool execution is prevented until policy checks are reachable again.

Integration Examples

LangChain AgentExecutor

Thoth detects AgentExecutor automatically and wraps both tool.run and tool._run:

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
import thoth

@tool
def web_search(query: str) -> str:
    """Search the web for current information."""
    ...

@tool
def read_file(path: str) -> str:
    """Read a file from the local filesystem."""
    ...

llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_tools_agent(llm, tools=[web_search, read_file], prompt=...)
executor = AgentExecutor(agent=agent, tools=[web_search, read_file])

# One call instruments all tools
executor = thoth.instrument(
    executor,
    agent_id="research-agent",
    approved_scope=["web_search", "read_file"],
    tenant_id="acme-corp",
    user_id="bob@acme.com",
    enforcement="block",
)

# Now every tool invocation is policy-checked
result = executor.invoke({"input": "Find recent SEC filings for AAPL"})

LangGraph

For LangGraph agents, wrap tool callables directly using Tracer.wrap_tool. Because LangGraph doesn't use AgentExecutor, the low-level API is needed here:

import os
import thoth
from thoth.models import ThothConfig, EnforcementMode
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
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_database(query: str) -> str:
    """Search the internal knowledge base."""
    ...

@tool
def send_slack_message(channel: str, message: str) -> str:
    """Post a message to Slack."""
    ...

api_key = os.environ["THOTH_API_KEY"]
config = ThothConfig(
    agent_id="langgraph-analyst",
    approved_scope=["search_database", "send_slack_message"],
    tenant_id=os.environ["THOTH_TENANT_ID"],
    user_id="alice@acme.com",
    enforcement=EnforcementMode.BLOCK,
    api_key=api_key,
)
session = SessionContext(config)
tracer = Tracer(
    config=config,
    session=session,
    emitter=HttpEmitter(api_url=config.resolved_api_url, api_key=api_key),
    enforcer=EnforcerClient(config),
    step_up=StepUpClient(config),
)

governed_search = tracer.wrap_tool("search_database", search_database)
governed_slack = tracer.wrap_tool("send_slack_message", send_slack_message)

llm = ChatAnthropic(model="claude-sonnet-4-6")
agent = create_react_agent(llm, [governed_search, governed_slack])

result = agent.invoke({"messages": [("user", "Search for Q4 data and post it to #analytics")]})

See examples/langgraph_example.py for the full working script.

Claude Agent SDK (query)

For claude-agent-sdk, instrument ClaudeAgentOptions instead of a .tools list. Thoth hooks into can_use_tool and tool lifecycle hooks.

import anyio
import os
from claude_agent_sdk import ClaudeAgentOptions, query
import thoth


async def prompt_stream():
    # can_use_tool callbacks require streaming mode in claude-agent-sdk
    yield {
        "type": "user",
        "message": {"role": "user", "content": "Read README.md and summarize risks."},
        "parent_tool_use_id": None,
        "session_id": "thoth-demo-session",
    }


async def main():
    options = ClaudeAgentOptions(
        max_turns=1,
        allowed_tools=["Read"],       # auto-allow list in Claude SDK
        disallowed_tools=["Bash"],    # explicit deny list in Claude SDK
    )

    options = thoth.instrument_claude_agent_sdk(
        options,
        agent_id="claude-agent-sdk-demo",
        approved_scope=["Read"],      # Thoth policy scope
        tenant_id=os.environ["THOTH_TENANT_ID"],
        user_id="alice@acme.com",
        enforcement="block",
    )

    async for message in query(prompt=prompt_stream(), options=options):
        print(message)


anyio.run(main)

See examples/claude_agent_sdk_example.py for the full working script.

Anthropic Claude

Use instrument_anthropic() to wrap tool functions for the Anthropic agentic loop:

import os
import anthropic
import thoth
from thoth import ThothPolicyViolation

governed = thoth.instrument_anthropic(
    {"search_docs": search_docs, "delete_record": delete_record},
    agent_id="claude-research-agent",
    approved_scope=["search_docs"],
    tenant_id=os.environ["THOTH_TENANT_ID"],
    user_id="alice@acme.com",
    enforcement="step_up",
    # THOTH_API_KEY read from env automatically
)

client = anthropic.Anthropic()
messages = [{"role": "user", "content": "Find our data retention policy."}]

while True:
    response = client.messages.create(
        model="claude-opus-4-6", max_tokens=1024, tools=TOOLS, messages=messages
    )
    if response.stop_reason == "end_turn":
        break
    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            try:
                result = governed[block.name](block.input)   # Thoth runs here
            except ThothPolicyViolation as e:
                result = f"[blocked: {e.reason}]"
            tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(result)})
    messages += [{"role": "assistant", "content": response.content},
                 {"role": "user", "content": tool_results}]

See examples/anthropic_example.py for the full working script.

OpenAI Function Calling

Use instrument_openai() to wrap tool functions for the OpenAI tool-calling loop:

import os
import openai
import thoth
from thoth import ThothPolicyViolation

def search_docs(query: str) -> str:
    ...

def send_email(to: str, subject: str, body: str) -> str:
    ...

governed = thoth.instrument_openai(
    {"search_docs": search_docs, "send_email": send_email},
    agent_id="openai-agent",
    approved_scope=["search_docs"],
    tenant_id=os.environ["THOTH_TENANT_ID"],
    user_id="charlie@acme.com",
    enforcement="block",
    # THOTH_API_KEY read from env automatically
)

client = openai.OpenAI()
messages = [{"role": "user", "content": "Find docs about access control"}]

while True:
    response = client.chat.completions.create(model="gpt-4o", tools=TOOLS, messages=messages)
    msg = response.choices[0].message
    if not msg.tool_calls:
        break
    messages.append(msg)
    for call in msg.tool_calls:
        try:
            import json
            result = governed[call.function.name](json.loads(call.function.arguments))
        except ThothPolicyViolation as e:
            result = f"[blocked: {e.reason}]"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})

See examples/openai_example.py for the full working script.

CrewAI

thoth.instrument() detects CrewAI Agent objects automatically:

import os
import thoth
from crewai import Agent
from crewai.tools import tool

@tool("web_search")
def web_search(query: str) -> str:
    """Search the web."""
    ...

researcher = Agent(role="Researcher", goal="Find information", tools=[web_search])

thoth.instrument(
    researcher,
    agent_id="crewai-researcher",
    approved_scope=["web_search"],
    tenant_id=os.environ["THOTH_TENANT_ID"],
    user_id="alice@acme.com",
    enforcement="block",
    # THOTH_API_KEY read from env automatically
)

See examples/crewai_example.py for the full working script.

AutoGen

Use one-call toolchain instrumentation and let the SDK build AutoGen's function_map for you:

import os
import autogen
import thoth

class AutoGenToolchain:
    def search_docs(self, query: str) -> str:
        return f"[docs results for '{query}']"

    def send_email(self, to: str, subject: str, body: str) -> str:
        return f"Email sent to {to}"

governed = thoth.instrument_toolchain(
    AutoGenToolchain(),
    agent_id="autogen-assistant",
    approved_scope=["search_docs", "send_email"],
    tenant_id=os.environ["THOTH_TENANT_ID"],
    enforcement="block",
)

function_map = thoth.toolchain_function_map(governed)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    function_map=function_map,  # built automatically from governed toolchain
)

See examples/autogen_example.py for the full working script.

Custom Tools (Generic)

Any object with a .tools list and tools that have a .run method or are directly callable will be instrumented automatically:

import thoth
from thoth import ThothPolicyViolation

class FileTool:
    name = "read_file"

    def run(self, path: str) -> str:
        with open(path) as f:
            return f.read()

class MyAgent:
    tools = [FileTool()]

    def run(self, prompt: str) -> str:
        # ... your agent logic calls self.tools[0].run(...)
        ...

agent = MyAgent()

agent = thoth.instrument(
    agent,
    agent_id="file-agent",
    approved_scope=["read_file"],
    tenant_id="acme-corp",
    enforcement="block",
)

try:
    result = agent.tools[0].run("/etc/passwd")
except ThothPolicyViolation as e:
    print(f"Blocked: {e.tool_name} — {e.reason}")
    if e.violation_id:
        print(f"Violation record: {e.violation_id}")

For nested toolchains (dicts + helper classes), use a single call to recursively instrument public methods and nested callables:

import thoth

class DataSources:
    def cloudtrail(self, args: dict) -> list[dict]:
        return self._fetch_cloudtrail(args["signal_id"])

    def _fetch_cloudtrail(self, signal_id: str) -> list[dict]:
        return [{"signal_id": signal_id}]

toolchain = {
    "data_sources": DataSources(),
    "archive_signal": lambda args: True,
}

governed = thoth.instrument_toolchain(
    toolchain,
    agent_id="alert-triage-agent",
    approved_scope=["data_sources.cloudtrail", "archive_signal"],
    tenant_id="acme-corp",
    enforcement="observe",
)

events = governed["data_sources"].cloudtrail({"signal_id": "sig-123"})

For runtimes that still require a function-map boundary (for example AutoGen), generate it from the governed toolchain automatically:

function_map = thoth.toolchain_function_map(governed)

Enforcement Modes

Set via the enforcement parameter to instrument().

Policy Decisions

The enforcer returns one of five decisions for each tool call:

By default, enforcer errors (for example, network timeout or 5xx) fail closed: the SDK treats them as BLOCK and raises ThothPolicyViolation. Set fail_open=True (or THOTH_FAIL_OPEN=true) to allow on transport/availability failures (network, 429, 5xx). Authentication/ingress failures remain blocked and include diagnostic hints in logs.

Handling Violations

from thoth import ThothPolicyViolation

try:
    result = agent.tools[0].run(user_input)
except ThothPolicyViolation as e:
    # e.tool_name   — the tool that was blocked
    # e.reason      — human-readable policy reason
    # e.violation_id — reference ID for the violation record in Maat dashboard
    # e.decision_reason_code / e.action_classification / e.authorization_decision
    # e.risk_score / e.pack_id / e.pack_version / e.rule_version
    # e.matched_rule_ids / e.matched_control_ids / e.regulatory_regimes
    # e.policy_references / e.model_signals
    # e.fastml_features / e.score_components / e.top_contributors / e.decision_evidence
    # e.enforcement_trace_id / e.receipt
    logger.warning("Policy violation on %s: %s (id=%s)", e.tool_name, e.reason, e.violation_id)
    return {"error": "This action is not permitted under your current access policy."}

Step-Up Authentication

When the enforcer returns STEP_UP, Thoth automatically:

1. Sends an approval request to the configured notification channel (Slack by default).
2. Polls /v1/enforce/hold/{hold_token} every step_up_poll_interval_seconds (default: 5s).
3. If approved within step_up_timeout_minutes (default: 15 minutes) → tool executes.
4. If timed out → raises ThothPolicyViolation with reason "step-up auth timeout".

If the response has no hold token, or approval remains unresolved after polling, the SDK raises ThothPolicyViolation and does not execute the tool. This applies to both synchronous and asynchronous wrapped tools. The Claude Agent SDK adapter returns a permission denial for the same conditions.

The Claude adapter carries the provider's tool-use ID through authorization and tool lifecycle events, unless action_attestation_id is explicitly configured. These IDs correlate events; they do not establish authenticated workload identity. When a lifecycle hook has neither ID, it omits the action ID rather than guessing which concurrent call it belongs to.

Configure timeouts via ThothConfig:

from thoth.models import ThothConfig, EnforcementMode

config = ThothConfig(
    agent_id="sensitive-agent",
    approved_scope=["delete_record"],
    tenant_id="acme-corp",
    enforcement=EnforcementMode.STEP_UP,
    step_up_timeout_minutes=5,       # fail fast in production
    step_up_poll_interval_seconds=3,
)

Session Inspection

Each instrument() call creates a SessionContext stored in a contextvars.ContextVar. Inspect it from anywhere in the same async context:

from thoth import get_current_session

session = get_current_session()
if session:
    print(f"Session ID:    {session.session_id}")
    print(f"Tools called:  {session.tool_calls}")
    print(f"Token spend:   {session.token_spend}")
    print(f"In scope:      {session.is_in_scope('web_search')}")

Configuration Reference

instrument() parameters

approved_scope clarifier:

• This field is an allow-list, not a natural-language description field.
• Entries must match emitted tool names exactly (for example search_docs or data_sources.cloudtrail).
• Tool meaning/context should be expressed through your tool schemas/prompts and policy context fields (session_intent, purpose, data_classification, task_context), which are sent to the enforcer.

instrument_toolchain() uses the same governance parameters plus:

ThothConfig fields

Environment Variables

Resolution precedence:

• THOTH_API_URL overrides api_url.

Dashboard

View sessions, violations, step-up requests, and policy decisions in your Thoth dashboard at https://<tenant>.<apex-domain>.

The dashboard shows:

• Sessions — per-agent session timelines with all tool calls
• Violations — blocked or escalated actions with full context
• Approvals — step-up queue for human reviewers
• Policies — view and edit the rules driving enforcement decisions
• Behavioral Analytics — drift detection and anomaly scores over time

License

Copyright 2026 Aten Security, Inc. All rights reserved.