tenuo 0.1.0b23

Capability tokens for AI agents - Python SDK

Tenuo Python SDK

Capability tokens for AI agents

Status: v0.1 Beta - 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 / LangGraph
uv pip install "tenuo[crewai]"        # + CrewAI
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)
uv pip install "tenuo[temporal]"      # + Temporal Python SDK (workflow + activity authorization)

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
bound = warrant.bind(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}
pop_sig = warrant.sign(agent_key, "manage_infrastructure", args)
authorized = warrant.authorize(
    tool="manage_infrastructure",
    args=args,
    signature=bytes(pop_sig)
)

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()

Key Management

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
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
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:

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

# 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 [e-mail]"}],
    tools=[...]
)  # Blocked: to doesn't match *@company.com

Security Constraints

Constraint	Purpose	Example
Subpath(root)	Blocks path traversal attacks	Subpath("/data") blocks /data/../etc/passwd
UrlSafe()	Blocks SSRF (private IPs, metadata)	UrlSafe() blocks http://169.254.169.254/
Shlex(allow)	Blocks shell injection	Shlex(allow=["ls"]) blocks ls; rm -rf /
Pattern(glob)	Glob pattern matching	Pattern("*@company.com")
UrlPattern(url)	URL matching. Note: https://example.com/ (trailing slash) parses as Wildcard ("Any Path"). Use /* to restrict to root.	UrlPattern("https://*.example.com/*")

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())

# Protect an entire crew
protected_crew = guard.protect(crew)
result = protected_crew.kickoff()

# Or use with Flows
from tenuo.crewai import guarded_tool

@guarded_tool(path=Subpath("/data"))
def read_file(path: str) -> str:
    return open(path).read()

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"

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(
    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

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):

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
warrant.is_expired     # bool
warrant.is_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.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,       # send warrant via params._meta.tenuo
) as client:
    ...

Server — verify warrants inside MCP tool handlers:

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

verifier = MCPVerifier(
    authorizer=Authorizer(trusted_roots=[PublicKey.from_bytes(root_pub)]),
    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()

Guard-protected tools return JSON-RPC error -32002 when approval is required. See examples/mcp/ for complete examples.

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:

# '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

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
• Temporal - Workflow + activity authorization (replay-safe)
• Security - Threat model, best practices
• API Reference - Full SDK docs

License

MIT OR Apache-2.0