Tenuo Python SDK

Capability tokens for AI agents

Status: v0.2 - Production/Stable. Core semantics are stable. See CHANGELOG.

Python bindings for Tenuo, providing cryptographically-enforced capability attenuation for AI agent workflows.

Installation

uv pip install tenuo                  # Core only
uv pip install "tenuo[openai]"        # + OpenAI Agents SDK
uv pip install "tenuo[google_adk]"    # + Google ADK
uv pip install "tenuo[a2a]"           # + Agent-to-Agent (inter-agent delegation)
uv pip install "tenuo[langchain]"     # + LangChain
uv pip install "tenuo[langgraph]"     # + LangGraph (includes LangChain)
uv pip install "tenuo[crewai]"        # + CrewAI (Python ≥3.10)
uv pip install "tenuo[autogen]"       # + AutoGen AgentChat (Python ≥3.10)
uv pip install "tenuo[fastapi]"       # + FastAPI
uv pip install "tenuo[mcp]"           # + official MCP SDK, client/server (Python ≥3.10)
uv pip install "tenuo[fastmcp]"       # + FastMCP (``TenuoMiddleware``, FastMCP servers; Python ≥3.10)
uv pip install "tenuo[temporal]"      # + Temporal Python SDK (workflow + activity authorization)
uv pip install "tenuo[cloud]"         # optional: proprietary Tenuo Cloud client (see below)

To have a supported coding agent inspect the installed SDK and implement a tested effect boundary for this project:

npx skills add tenuo-ai/tenuo --skill tenuo-agent-authorization

The skill labels official adapters, verified recipes, generic integrations, and unsupported framework paths separately.

Tenuo Cloud (optional)

The open-source tenuo package implements the protocol: warrants, constraints, proof-of-possession, local enforcement, and framework hooks. It runs fully self-hosted with your own keys and authorizer. No Cloud account required.

Tenuo Cloud (currently in early access) is an optional managed control plane on top of that protocol (KMS, agent registry, warrant pipeline, signed revocation lists, signed receipts, discovery, human approvals, REST API). Any language can use Cloud via HTTP; the proprietary Python SDK is not required.

Combine extras as needed, e.g. pip install "tenuo[cloud,temporal]" or "tenuo[cloud,langgraph]".

For Python Cloud integration (credentials, connect() / doctor(), import swaps, approval gates), see the tenuo-cloud PyPI README and docs.tenuo.ai; those guides are not part of this open-source repo.

Claude Code governance uses tenuo-claude-code against the same Cloud tenant, not tenuo-cloud. See Claude Code Governance.

Development

We recommend using uv for development. It manages Python versions and dependencies deterministically.

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Sync environment (creates .venv and installs dependencies)
uv sync --all-extras

You can still use standard pip if you prefer:

python -m venv .venv
source .venv/bin/activate
uv pip install -e ".[dev]"

Quick Start

30-Second Demo (Copy-Paste)

from tenuo import configure, SigningKey, mint_sync, guard, Capability, Pattern

configure(issuer_key=SigningKey.generate(), dev_mode=True, audit_log=False)

@guard(tool="search")
def search(query: str) -> str:
    return f"Results for: {query}"

with mint_sync(Capability("search", query=Pattern("weather *"))):
    print(search(query="weather NYC"))   # OK: Results for: weather NYC
    print(search(query="stock prices"))  # Raises AuthorizationDenied

dev_mode=True above is for local development only: it relaxes trust-root and audit-log requirements so the snippet runs out of the box. The next section shows the production pattern.

The Safe Path (Production Pattern)

In production, you receive warrants from an orchestrator and keep keys separate:

from tenuo import Warrant, SigningKey, Pattern

# In production: receive warrant as base64 string from orchestrator
# warrant = Warrant(received_warrant_string)

# For testing: create one yourself
key = SigningKey.generate()
warrant = (Warrant.mint_builder()
    .tool("search")
    .holder(key.public_key)
    .ttl(3600)
    .mint(key))

# Explicit key at call site - keys never in state
headers = warrant.headers(key, "search", {"query": "test"})

# Delegation with attenuation
worker_key = SigningKey.generate()
child = warrant.grant(
    to=worker_key.public_key,
    allow="search",
    query=Pattern("safe*"),
    ttl=300,
    key=key
)

BoundWarrant (For Repeated Operations)

When you need to make many calls with the same warrant+key:

from tenuo import Warrant, SigningKey

# Create a warrant (in production: Warrant(received_base64_string))
key = SigningKey.generate()
warrant = (Warrant.mint_builder()
    .tool("process")
    .holder(key.public_key)
    .ttl(3600)
    .mint(key))

# Bind key for repeated use. trusted_roots is the anchor the warrant's issuer
# must chain back to; this warrant is self-minted, so that anchor is `key`.
bound = warrant.bind(key, trusted_roots=[key.public_key])

items = ["item1", "item2", "item3"]
for item in items:
    headers = bound.headers("process", {"item": item})
    # Make API call with headers...

# Validate before use
result = bound.validate("process", {"item": "test"})
if result:
    print("Authorized!")

# Note: BoundWarrant is non-serializable (contains key)
# Use bound.warrant to get the plain Warrant for storage

Low-Level API (Full Control)

# ┌─────────────────────────────────────────────────────────────────┐
# │  CONTROL PLANE / ORCHESTRATOR                                   │
# │  Issues warrants to agents. Only needs agent's PUBLIC key.      │
# └─────────────────────────────────────────────────────────────────┘
from tenuo import SigningKey, Warrant, Pattern, Range, PublicKey

issuer_key = SigningKey.from_env("ISSUER_KEY")
agent_pubkey = PublicKey.from_env("AGENT_PUBKEY")  # From registration

warrant = (Warrant.mint_builder()
    .capability("manage_infrastructure",
        cluster=Pattern("staging-*"),
        replicas=Range.max_value(15))
    .holder(agent_pubkey)
    .ttl(3600)
    .mint(issuer_key))

# Send warrant to agent: send_to_agent(str(warrant))

# ┌─────────────────────────────────────────────────────────────────┐
# │  AGENT / WORKER                                                 │
# │  Receives warrant, uses own private key for Proof-of-Possession │
# └─────────────────────────────────────────────────────────────────┘
from tenuo import SigningKey, Warrant

agent_key = SigningKey.from_env("AGENT_KEY")  # Agent's private key (never shared)
warrant = Warrant(received_warrant_string)    # Deserialize from orchestrator

args = {"cluster": "staging-web", "replicas": 5}

# Produce transport headers with the warrant and holder proof.
headers = warrant.headers(agent_key, "manage_infrastructure", args)

# Or validate locally before executing.
result = warrant.validate(agent_key, "manage_infrastructure", args)
if result:
    run_infrastructure_action(args)

Holder Runtime

Runtime owns identity, trusted roots, the current signed revocation list, and an optional receipt outbox so application code does not coordinate those pieces by hand.

from tenuo import HolderIdentity, Runtime

identity = HolderIdentity.load_or_create("holder.key")
runtime = Runtime(
    identity=identity,
    trusted_roots=roots,
    revocation_list=signed_srl,
    receipts="collect",
)
session = runtime.session_from_wire(warrant)
with runtime.session_scope(session):
    protected_tool(...)

batch = runtime.peek_receipts()
# upload, then:
runtime.acknowledge_receipts(len(batch))

ConnectToken.parse decodes a tenuo_ct_… token. It does not read environment variables.

Key Management

Loading Keys

from tenuo import SigningKey

# From environment variable (auto-detects base64/hex)
key = SigningKey.from_env("TENUO_ROOT_KEY")

# From file (auto-detects format)
key = SigningKey.from_file("/run/secrets/tenuo-key")

# Generate new
key = SigningKey.generate()

Runtime Key Stores

KeyRegistry (Thread-Safe Singleton)

LangGraph checkpoints state to databases. Private keys in state = private keys in your database. KeyRegistry solves this by keeping keys in memory while only string IDs flow through state.

from tenuo import KeyRegistry, SigningKey

registry = KeyRegistry.get_instance()

# At startup: register keys (keys stay in memory)
registry.register("worker", SigningKey.from_env("WORKER_KEY"))
registry.register("orchestrator", SigningKey.from_env("ORCH_KEY"))

# In your code: lookup by ID (ID is just a string, safe to checkpoint)
key = registry.get("worker")

# Multi-tenant: namespace keys per tenant
registry.register("api", tenant_a_key, namespace="tenant-a")
registry.register("api", tenant_b_key, namespace="tenant-b")
key = registry.get("api", namespace="tenant-a")

Use cases:

• LangGraph: Keys never in state, checkpointing-safe
• Multi-tenant SaaS: Isolate keys per tenant with namespaces
• Service mesh: Different keys per downstream service
• Key rotation: Register both current and previous keys

Keyring (For Key Rotation)

from tenuo import Keyring, SigningKey

keyring = Keyring(
    root=SigningKey.from_env("CURRENT_KEY"),
    previous=[SigningKey.from_env("OLD_KEY")]
)

# All public keys for verification (current + previous)
all_pubkeys = keyring.all_public_keys

FastAPI Integration

from fastapi import FastAPI, Depends
from tenuo.fastapi import TenuoGuard, SecurityContext, configure_tenuo

app = FastAPI()
configure_tenuo(app, trusted_issuers=[issuer_pubkey])

@app.get("/search")
async def search(
    query: str,
    ctx: SecurityContext = Depends(TenuoGuard("search"))
):
    # ctx.warrant is verified
    # ctx.args contains extracted arguments
    return {"results": [...]}

LangChain Integration

from tenuo import Warrant, SigningKey
from tenuo.langchain import guard

# Create bound warrant
keypair = SigningKey.generate()  # In production: SigningKey.from_env("MY_KEY")
warrant = (Warrant.mint_builder()
    .tools(["search"])
    .mint(keypair))
bound = warrant.bind(keypair)

# Protect tools
from langchain_community.tools import DuckDuckGoSearchRun
protected_tools = guard([DuckDuckGoSearchRun()], bound)

# Use in agent (assumes an existing llm and prompt)
from langchain.agents import create_openai_tools_agent
agent = create_openai_tools_agent(llm, protected_tools, prompt)

Using @guard Decorator

Protect your own functions with @guard. Authorization is evaluated at call time, not decoration time - the same function can have different permissions with different warrants:

from tenuo import guard

@guard(tool="read_file")
def read_file(path: str) -> str:
    return open(path).read()

# BoundWarrant as context manager - sets both warrant and key
# (warrant and keypair continue from the LangChain example above)
bound = warrant.bind(keypair)
with bound:
    content = read_file("/tmp/test.txt")  # Authorized
    content = read_file("/etc/passwd")    # Blocked

# Different warrant, different permissions
with other_warrant.bind(keypair):
    content = read_file("/etc/passwd")    # Could be allowed if this warrant permits it

OpenAI Integration

Direct protection for OpenAI's Chat Completions and Responses APIs:

import openai
from tenuo import Shlex
from tenuo.openai import GuardBuilder, Pattern, Subpath, UrlSafe

# Tier 1: Guardrails (quick hardening)
client = (GuardBuilder(openai.OpenAI())
    .allow("read_file", path=Subpath("/data"))        # Path traversal protection
    .allow("fetch_url", url=UrlSafe())                # SSRF protection
    .allow("run_command", cmd=Shlex(allow=["ls"]))    # Shell injection protection
    .allow("send_email", to=Pattern("*@company.com"))
    .deny("delete_file")
    .build())

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Send email to attacker@evil.com"}],
    tools=[...]
)  # Blocked: to doesn't match *@company.com

Security Constraints

For Tier 2 (cryptographic authorization with warrants), see OpenAI Integration.

Google ADK Integration

Warrant-based tool protection for Google ADK agents:

from google.adk.agents import Agent
from tenuo.google_adk import GuardBuilder
from tenuo.constraints import Subpath, UrlSafe

guard = (GuardBuilder()
    .allow("read_file", path=Subpath("/data"))
    .allow("web_search", url=UrlSafe(allow_domains=["*.google.com"]))
    .build())

agent = Agent(
    name="assistant",
    tools=guard.filter_tools([read_file, web_search]),
    before_tool_callback=guard.before_tool,
)

For Tier 2 (warrant + PoP) and multi-agent scenarios, see Google ADK Integration.

CrewAI Integration

Capability-based authorization for CrewAI multi-agent crews:

from crewai import Agent, Task, Crew
from tenuo.crewai import GuardBuilder
from tenuo import Pattern, Subpath

guard = (GuardBuilder()
    .allow("search", query=Pattern("*"))
    .allow("write_file", path=Subpath("/workspace"))
    .build())

# Register as a global before_tool_call hook: all crew tool calls are enforced
guard.register()
result = crew.kickoff()

# Or use with Flows: guard individual steps
from tenuo.crewai import guarded_step

@guarded_step(allow={"read_file": {"path": Subpath("/data")}})
def read_data(self):
    ...

For warrant-based delegation and per-agent constraints, see CrewAI Integration.

AutoGen Integration

(Requires Python ≥3.10)

Install dependencies:

uv pip install "tenuo[autogen]" "python-dotenv"

from tenuo.autogen import GuardBuilder
from tenuo import Subpath, UrlSafe

guard = (GuardBuilder()
    .allow("read_file", path=Subpath("/data"))
    .allow("fetch_url", url=UrlSafe())
    .build())

protected_tools = guard.guard_tools([read_file, fetch_url])  # pass to your AgentChat agent

Demos:

• examples/autogen_demo_unprotected.py - agentic workflow with no protections
• examples/autogen_demo_protected_tools.py - guarded tools (URL allowlist + Subpath)
• examples/autogen_demo_protected_attenuation.py - per-agent attenuation + escalation block

Tip: these demos use python-dotenv to load OPENAI_API_KEY and set tool_choice="required" for deterministic tool calls.

A2A Integration (Multi-Agent)

Warrant-based authorization for agent-to-agent communication via automated CSR handshake:

from tenuo.a2a import A2AServerBuilder, A2AClient
from tenuo.constraints import UrlSafe

# Server: Define skills and decide what capabilities to grant
server = (A2AServerBuilder()
    .name("Research Agent")
    .url("https://research-agent.example.com")
    .key(my_public_key)
    .trust(orchestrator_key)
    .registration_handler(my_handler) # Enable CSR handshake
    .build())

@server.skill("search_papers", constraints={"sources": UrlSafe})
async def search_papers(query: str, sources: list[str]) -> list[dict]:
    return await do_search(query, sources)

# Client: Automatically fetch warrant and execute task
client = A2AClient("https://research-agent.example.com")
warrant = await client.request_warrant(signing_key=worker_key, capabilities={"search_papers": {}})
result = await client.send_task(
    "Find recent AI agents papers",   # message (required)
    skill="search_papers",
    arguments={"query": "AI Agents"},
    warrant=warrant,
    signing_key=worker_key
)

See A2A Integration for full documentation.

Temporal Integration

(Requires Python ≥3.10 and temporalio>=1.23.0)

Warrant-based authorization for Temporal workflows and activities. The plugin wires the client interceptor, worker interceptor, and sandboxed workflow runner (with PyO3 passthrough) in one step:

from temporalio.client import Client
from temporalio.worker import Worker
from tenuo import SigningKey
from tenuo.temporal import TenuoPluginConfig, EnvKeyResolver
from tenuo.temporal_plugin import TenuoTemporalPlugin

control_key = SigningKey.generate()

plugin = TenuoTemporalPlugin(
    TenuoPluginConfig(
        key_resolver=EnvKeyResolver(),
        trusted_roots=[control_key.public_key],
    )
)

client = await Client.connect("localhost:7233", plugins=[plugin])
worker = Worker(client, task_queue="my-queue", workflows=[MyWorkflow], activities=[...])

Every activity invocation is verified against the workflow's warrant + PoP signature; deterministic replay is preserved. See Temporal Integration for the full guide.

LangGraph Integration

Assumes an existing StateGraph named graph:

from tenuo import KeyRegistry
from tenuo.langgraph import guard_node, TenuoToolNode, load_tenuo_keys

# Load keys from TENUO_KEY_* environment variables
load_tenuo_keys()

# Wrap pure nodes
def my_agent(state):
    return {"messages": [...]}

graph.add_node("agent", guard_node(my_agent, key_id="worker"))
graph.add_node("tools", TenuoToolNode([search, calculator]))

# Run with warrant in state (str() returns base64)
state = {"warrant": str(warrant), "messages": [...]}
config = {"configurable": {"tenuo_key_id": "worker"}}
result = graph.invoke(state, config=config)

Conditional Logic Based on Permissions

Use @tenuo_node when your node needs to check what the warrant allows:

from tenuo.langgraph import tenuo_node
from tenuo import BoundWarrant

@tenuo_node
def smart_router(state, bound_warrant: BoundWarrant):
    # Route based on what the warrant permits
    if bound_warrant.allows("search"):
        return {"next": "researcher"}
    return {"next": "fallback"}

Audit Logging

Tenuo logs all authorization events as JSON for observability:

{"event_type": "authorization_success", "tool": "search", "action": "authorized", ...}
{"event_type": "authorization_failure", "tool": "delete", "error_code": "CONSTRAINT_VIOLATION", ...}

To suppress logs (for testing/demos):

from tenuo import configure

configure(issuer_key=key, dev_mode=True, audit_log=False)

Or configure the audit logger directly:

from tenuo.audit import audit_logger
audit_logger.configure(enabled=False)  # Disable
audit_logger.configure(use_python_logging=True, logger_name="tenuo")  # Use Python logging

Debugging

why_denied() - Understand Failures

result = warrant.why_denied("read_file", {"path": "/etc/passwd"})
if result.denied:
    print(f"Code: {result.deny_code}")
    print(f"Field: {result.field}")
    print(f"Suggestion: {result.suggestion}")

diagnose() - Inspect Warrants

from tenuo import diagnose

diagnose(warrant)
# Prints: ID, TTL, constraints, tools, etc.

Convenience Properties

# Time remaining
warrant.ttl_remaining  # timedelta
warrant.ttl            # alias for ttl_remaining

# Status (properties; method forms is_expired() / is_terminal() also exist)
warrant.expired        # bool
warrant.terminal       # bool (can't delegate further)

# Human-readable
warrant.capabilities   # dict of tool -> constraints

MCP Integration

(Requires Python ≥3.10)

Client: connect to any MCP server with automatic warrant injection:

from tenuo import mint, Capability, Subpath
from tenuo.mcp import SecureMCPClient

# Stdio (local subprocess)
async with SecureMCPClient("python", ["server.py"]) as client:
    async with mint(Capability("read_file", path=Subpath("/data"))):
        result = await client.tools["read_file"](path="/data/file.txt")

# SSE or StreamableHTTP (remote server)
async with SecureMCPClient(
    url="https://mcp.example.com/mcp",
    transport="http",          # or "sse"
    inject_warrant=True,       # params._meta.tenuo; use "argument" for _tenuo
) as client:
    ...

Server: verify warrants inside MCP tool handlers:

from tenuo import Authorizer, PublicKey, CompiledMcpConfig, McpConfig
from tenuo.mcp import MCPVerifier

root_pubkey = PublicKey.from_env("TENUO_ROOT_PUBKEY")  # control plane's public key

verifier = MCPVerifier(
    authorizer=Authorizer(trusted_roots=[root_pubkey]),
    config=CompiledMcpConfig.compile(McpConfig.from_file("mcp-config.yaml")),
)

@mcp.tool()
async def read_file(path: str, **kwargs) -> str:
    clean = verifier.verify_or_raise("read_file", {"path": path, **kwargs})
    return open(clean["path"]).read()

MCP servers return JSON-RPC -32002 when an approval gate fires or multi-sig threshold is not met; retry with _meta.tenuo.approvals. See Human Approvals.

Security Considerations

BoundWarrant Serialization

BoundWarrant contains a private key and cannot be serialized:

bound = warrant.bind(key)

# This raises TypeError - BoundWarrant contains private key
pickle.dumps(bound)
json.dumps(bound)

# Extract warrant for storage (str() returns base64)
state["warrant"] = str(bound.warrant)
# Reconstruct later with Warrant(string)

allows() vs validate()

# allows() = Logic Check (Math only)
# Good for UI logic, conditional routing, fail-fast
if bound.allows("delete"):
    show_delete_button()

if bound.allows("delete", {"target": "users"}):
    print("Deletion would be permitted by constraints")

# validate() = Full Security Check (Math + Crypto)
# Proves you hold the key and validates the PoP signature
result = bound.validate("delete", {"target": "users"})
if result:
    delete_database()
else:
    print(f"Failed: {result.reason}")

Error Details Not Exposed

Authorization errors are opaque by default:

# Client sees: "Authorization denied (ref: abc123)"
# Logs show: "[abc123] Constraint failed: path=/etc/passwd, expected=Pattern(/data/*)"

Closed-World Constraints

Once you add any constraint, unknown arguments are rejected. Inside a Warrant.mint_builder() chain:

# 'timeout' is unknown - blocked by closed-world policy
.capability("api_call", url=UrlSafe(allow_domains=["api.example.com"]))

# Use Wildcard() for specific fields you want to allow
.capability("api_call", url=UrlSafe(allow_domains=["api.example.com"]), timeout=Wildcard())

# Or opt out of closed-world entirely
.capability("api_call", url=UrlSafe(allow_domains=["api.example.com"]), _allow_unknown=True)

Examples

# Basic usage
python examples/basic_usage.py

# FastAPI integration
python examples/fastapi_integration.py

# LangGraph protected
python examples/langchain/langgraph_protected.py

# MCP integration
python examples/mcp/mcp_client_demo.py

Requirements

Documentation

• Quickstart - Get running in 5 minutes
• OpenAI - Direct API protection with streaming defense
• Google ADK - ADK agent tool protection
• AutoGen - AgentChat tool protection
• A2A - Inter-agent delegation with warrants
• FastAPI - Zero-boilerplate API protection
• LangChain - Tool protection
• LangGraph - Multi-agent security
• CrewAI - Multi-agent crew protection
• MCP - Model Context Protocol client + server verification
• Temporal - Workflow + activity authorization (replay-safe)
• Security - Threat model, best practices
• Tenuo Cloud - Optional managed control plane (API + proprietary tenuo-cloud SDK)
• API Reference - Full SDK docs

License

Apache-2.0. See LICENSE for details.