atensec-thoth 0.1.16

AI agent governance SDK by Aten Security

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="[e-mail]",
    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_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="[e-mail]",
    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="[e-mail]",
    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="[e-mail]",
        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="[e-mail]",
    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="[e-mail]",
    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="[e-mail]",
    enforcement="block",
    # THOTH_API_KEY read from env automatically
)

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

AutoGen

Use wrap_autogen_tools() to govern AutoGen's function_map:

import os
import autogen
from thoth.integrations.autogen import wrap_autogen_tools
from thoth.tracer import Tracer
# ... build tracer as shown in the LangGraph section above ...

governed = wrap_autogen_tools(
    {"search_docs": search_docs, "send_email": send_email},
    tracer=tracer,
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    function_map=governed,   # Thoth enforcement runs on every function call
)

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}")

---

Enforcement Modes

Set via the enforcement parameter to instrument().

Mode	Value	Behavior
Observe	observe	All tool calls pass through. Events are still emitted for audit. No blocking, no step-up. Use for initial rollout and baselining.
Step-Up	step_up	Suspicious calls trigger a human approval request (e.g. Slack DM to a reviewer). Tool execution is held until approved or timed out.
Block	block	Calls that violate policy raise ThothPolicyViolation immediately.
Progressive	progressive	Escalating mode. Enforcer chooses the appropriate response per tool call based on policy rules.

---

Policy Decisions

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

Decision	Meaning	Agent behavior
ALLOW	Call is within policy.	Tool executes immediately.
STEP_UP	Call requires human approval.	SDK polls /v1/enforce/hold/{token} until approved or timed out. On timeout → BLOCK.
MODIFY	Call is allowed with transformed/sanitized arguments.	SDK rewrites the tool call args and then executes the tool.
DEFER	Call is postponed pending more context/workflow state.	SDK raises ThothPolicyViolation with defer context and timeout guidance.
BLOCK	Call violates policy.	ThothPolicyViolation is raised before the tool executes.

Enforcer errors (for example, network timeout or 5xx) fail closed: the SDK treats them as BLOCK and raises ThothPolicyViolation. Authentication/ingress failures include diagnostic hints in logs to speed up misconfiguration debugging.

---

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".

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

Parameter	Type	Default	Description
agent	Any	—	Agent object to instrument. Must have a .tools list.
agent_id	str	—	Stable identifier for this agent definition. Used in policy rules.
approved_scope	list[str]	—	List of tool names this agent is authorized to call.
tenant_id	str	—	Your Maat tenant ID.
user_id	str	"system"	Identity of the user on whose behalf the agent acts.
enforcement	str	"block"	Enforcement mode: observe, step_up, block, or progressive.
api_key	str | None	$THOTH_API_KEY	API key from the Aten dashboard. Events sent over HTTPS — no AWS credentials required.
event_ingest_token	str | None	$THOTH_EVENT_INGEST_TOKEN	Optional dedicated token sent as X-Thoth-Event-Ingest-Token for /v1/events/batch.
api_url	str | None	$THOTH_API_URL	Required tenant API base URL used for both event ingestion and policy checks.
session_id	str | None	auto-generated UUID	Pass an existing session ID to continue a session across calls.
session_intent	str | None	None	Intent label used for minimum-necessary/session-scope policies.
purpose	str | None	None	Policy context for allowed business purpose.
data_classification	str | None	None	Policy context for data sensitivity.
task_context	dict[str, Any] | None	{}	Optional delegation metadata (initiated_by, task_id, chain, etc.).
environment	str | None	$THOTH_ENVIRONMENT or "prod"	Environment selector sent to enforcer policy lookup.
enforcement_trace_id	str | None	session_id	Correlation ID propagated across enforcement/evidence paths.

ThothConfig fields

Field	Type	Default	Description
agent_id	str	—	Stable identifier for this agent.
approved_scope	list[str]	—	Tool names the agent is authorized to call.
tenant_id	str	—	Your Maat tenant ID.
user_id	str	"system"	User identity for the session.
enforcement	EnforcementMode	BLOCK	Enforcement mode.
api_key	str | None	None	Aten API key. Falls back to THOTH_API_KEY env var via _build_components.
event_ingest_token	str | None	None	Optional dedicated token for telemetry ingest. Falls back to THOTH_EVENT_INGEST_TOKEN.
api_url	str | None	None	Required tenant API base URL. Provide directly or via THOTH_API_URL. Used for both /v1/enforce and /v1/events/batch.
step_up_timeout_minutes	int	15	Timeout for step-up approval.
step_up_poll_interval_seconds	int	5	Polling interval for step-up hold status.
session_intent	str | None	None	Session purpose used by intent-scoped compliance packs.
purpose	str | None	None	Business-purpose label propagated to policy checks and telemetry.
data_classification	str | None	None	Sensitivity label propagated to policy checks and telemetry.
task_context	dict[str, Any]	{}	Delegation/task metadata included in enforcement payload + events.
environment	str	"prod"	Environment selector included in /v1/enforce payload.
enforcement_trace_id	str | None	None	Optional explicit trace ID; falls back to session_id.

Environment Variables

Variable	Description
THOTH_TENANT_ID	Tenant identifier used by your application to pass tenant_id into SDK APIs (the SDK does not auto-read this by itself).
THOTH_API_KEY	API key from the Aten dashboard. Enables HTTPS event transport. Example: thoth_live_abc123...
THOTH_EVENT_INGEST_TOKEN	Optional dedicated telemetry token for /v1/events/batch (X-Thoth-Event-Ingest-Token header).
THOTH_API_URL	Required tenant API base URL for both event ingestion and policy checks. Example: https://enforce.<tenant>.<apex-domain>
THOTH_ENVIRONMENT	Optional environment selector (prod, staging, dev, etc.) sent to the enforcer payload.
THOTH_LOG_LEVEL	Optional SDK logger level override (DEBUG, INFO, WARNING, ERROR). If unset, SDK falls back to LOG_LEVEL when present.

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.