Metadata-Version: 2.4
Name: tenuo
Version: 0.3.2
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Rust
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Security
Classifier: Topic :: Security :: Cryptography
Classifier: Topic :: Software Development :: Libraries
Classifier: Framework :: FastAPI
Classifier: Typing :: Typed
Requires-Dist: pyyaml>=6.0
Requires-Dist: pydantic>=2.0
Requires-Dist: httpx>=0.24 ; extra == 'a2a'
Requires-Dist: starlette>=0.27 ; extra == 'a2a'
Requires-Dist: autogen-agentchat>=0.7.0 ; python_full_version >= '3.10' and extra == 'autogen'
Requires-Dist: autogen-ext[openai]>=0.7.0 ; python_full_version >= '3.10' and extra == 'autogen'
Requires-Dist: rich>=10.0.0 ; extra == 'cli'
Requires-Dist: tenuo-cloud>=0.2,<0.3 ; extra == 'cloud'
Requires-Dist: crewai>=1.5 ; python_full_version >= '3.10' and extra == 'crewai'
Requires-Dist: maturin>=1.9.3 ; extra == 'dev'
Requires-Dist: pytest>=7.0 ; extra == 'dev'
Requires-Dist: pytest-asyncio ; extra == 'dev'
Requires-Dist: hypothesis>=6.0 ; extra == 'dev'
Requires-Dist: ruff>=0.1.0 ; extra == 'dev'
Requires-Dist: mypy>=1.0 ; extra == 'dev'
Requires-Dist: types-pyyaml ; extra == 'dev'
Requires-Dist: types-requests ; extra == 'dev'
Requires-Dist: uvicorn ; extra == 'dev'
Requires-Dist: langchain-openai ; extra == 'dev'
Requires-Dist: tenuo[openai,autogen,langchain,langgraph,crewai,a2a,google-adk,fastapi,mcp,fastmcp,temporal] ; extra == 'dev'
Requires-Dist: fastapi>=0.100.0 ; extra == 'fastapi'
Requires-Dist: uvicorn>=0.20.0 ; extra == 'fastapi'
Requires-Dist: mcp>=1.9.4 ; python_full_version >= '3.10' and extra == 'fastmcp'
Requires-Dist: fastmcp>=3.2.1 ; python_full_version >= '3.10' and extra == 'fastmcp'
Requires-Dist: google-adk>=0.1 ; extra == 'google-adk'
Requires-Dist: langchain-core>=0.2 ; extra == 'langchain'
Requires-Dist: langchain>=0.2 ; extra == 'langchain'
Requires-Dist: langgraph>=0.2 ; extra == 'langgraph'
Requires-Dist: langchain-core>=0.2.27 ; extra == 'langgraph'
Requires-Dist: mcp>=1.9.4 ; python_full_version >= '3.10' and extra == 'mcp'
Requires-Dist: openai>=1.0 ; extra == 'openai'
Requires-Dist: openai-agents>=0.1 ; python_full_version >= '3.10' and extra == 'openai'
Requires-Dist: temporalio>=1.23.0 ; python_full_version >= '3.10' and extra == 'temporal'
Requires-Dist: boto3>=1.26.0 ; extra == 'temporal'
Requires-Dist: google-cloud-secret-manager>=2.16.0 ; extra == 'temporal'
Provides-Extra: a2a
Provides-Extra: autogen
Provides-Extra: cli
Provides-Extra: cloud
Provides-Extra: crewai
Provides-Extra: dev
Provides-Extra: fastapi
Provides-Extra: fastmcp
Provides-Extra: google_adk
Provides-Extra: langchain
Provides-Extra: langgraph
Provides-Extra: mcp
Provides-Extra: openai
Provides-Extra: temporal
License-File: LICENSE
License-File: NOTICE
Summary: Task-scoped authorization for AI agents: signed warrants limit which tools an agent can call, with which arguments, and for how long. Adapters for LangChain, LangGraph, CrewAI, OpenAI Agents SDK, Google ADK, AutoGen, MCP, FastMCP, A2A, FastAPI, and Temporal.
Keywords: authorization,ai-agents,agent-security,llm-security,access-control,least-privilege,capabilities,warrants,delegation,prompt-injection,mcp,fastmcp,model-context-protocol,langchain,langgraph,crewai,openai-agents,google-adk,autogen,a2a,temporal,fastapi
Author: Tenuo Contributors
License-Expression: Apache-2.0
Requires-Python: >=3.9
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Changelog, https://github.com/tenuo-ai/tenuo/blob/main/CHANGELOG.md
Project-URL: Documentation, https://tenuo.ai/quickstart
Project-URL: Homepage, https://tenuo.ai
Project-URL: Issues, https://github.com/tenuo-ai/tenuo/issues
Project-URL: Repository, https://github.com/tenuo-ai/tenuo
Project-URL: Security Policy, https://github.com/tenuo-ai/tenuo/blob/main/SECURITY.md

# Tenuo Python SDK

**Task-scoped authorization for AI agents**

[![PyPI](https://img.shields.io/pypi/v/tenuo.svg)](https://pypi.org/project/tenuo/)
[![Python Versions](https://img.shields.io/pypi/pyversions/tenuo.svg)](https://pypi.org/project/tenuo/)

> **Status: v0.2 - Production/Stable.** Core semantics are stable. See [CHANGELOG](https://github.com/tenuo-ai/tenuo/blob/main/CHANGELOG.md).

The Python SDK for [Tenuo](https://github.com/tenuo-ai/tenuo). A warrant is a signed grant of which tools an agent may call, with which argument constraints, and for how long. Tenuo checks every tool call against it before the tool runs, so a prompt-injected or confused agent cannot act outside its task. Warrants are bound to the agent holding them, can only narrow when delegated to another agent, verify offline, and produce signed allow/deny receipts. Adapters cover LangChain, LangGraph, CrewAI, the OpenAI Agents SDK, Google ADK, AutoGen, MCP (official SDK and FastMCP), A2A, FastAPI, and Temporal.

## Installation

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

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

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

[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/tenuo-ai/tenuo/blob/main/notebooks/tenuo_demo.ipynb)
[![Explorer](https://img.shields.io/badge/Explorer-decode_warrants-1a1a1a)](https://tenuo.ai/explorer/)

## 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**](https://cloud.tenuo.ai) (currently in [early access](https://tenuo.ai/early-access.html)) 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.

| Path | Install | Details |
|------|---------|---------|
| **Self-host** | `pip install tenuo` | Local enforcement; you operate keys and issuance |
| **Cloud, no SDK** | `pip install tenuo` + `TENUO_API_KEY` | Call the [REST API](https://docs.tenuo.ai/api-reference) yourself (`httpx`, `curl`, etc.) |
| **Cloud + SDK** | `pip install "tenuo[cloud]"` | Pulls [`tenuo-cloud`](https://pypi.org/project/tenuo-cloud/) (proprietary wheels): Quick Connect, `cloud_approval()`, framework drop-ins |

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](https://pypi.org/project/tenuo-cloud/)** and **[docs.tenuo.ai](https://docs.tenuo.ai)**; those guides are not part of this open-source repo.

**Claude Code** governance uses [`tenuo-claude-code`](https://pypi.org/project/tenuo-claude-code/) against the same Cloud tenant, not `tenuo-cloud`. See [Claude Code Governance](https://docs.tenuo.ai/guides/claude-code).

## Development

We recommend using [uv](https://github.com/astral-sh/uv) for development. It manages Python versions and dependencies deterministically.

```bash
# 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:

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

## Quick Start

### 30-Second Demo (Copy-Paste)

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

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

```python
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)

```python
# ┌─────────────────────────────────────────────────────────────────┐
# │  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))
```

```python
# ┌─────────────────────────────────────────────────────────────────┐
# │  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.

```python
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

```python
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.

```python
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)

```python
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

```python
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

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

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

```python
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

| 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/*")` |

Tier 1 rejects out-of-policy calls on the guarded path. The constraints above are that policy: path scope, URL safety, shell allowlists, and patterns. Tier 2 carries the same policy in a signed, holder-bound warrant, with delegation that can only narrow. See [OpenAI Integration](https://tenuo.ai/openai). When the agent can skip the wrapper, enforce in the component that performs the effect, outside the agent's control.

## Google ADK Integration

Local policy checks for Google ADK agents (Tier 1):

```python
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](https://tenuo.ai/google-adk).

## CrewAI Integration

Capability-based authorization for CrewAI multi-agent crews:

```python
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](https://tenuo.ai/crewai).

## AutoGen Integration

_(Requires Python ≥3.10)_

Install dependencies:

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

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

```python
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](https://tenuo.ai/a2a) 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:

```python
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](https://tenuo.ai/temporal) for the full guide.

## LangGraph Integration

Assumes an existing `StateGraph` named `graph`:

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

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

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

To suppress logs (for testing/demos):

```python
from tenuo import configure

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

Or configure the audit logger directly:

```python
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

```python
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

```python
from tenuo import diagnose

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

### Convenience Properties

```python
# 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:

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

```python
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](https://tenuo.ai/approvals).

## Security Considerations

### BoundWarrant Serialization

`BoundWarrant` contains a private key and **cannot be serialized**:

```python
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()`
 
 ```python
 # 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:

```python
# 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:

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

```bash
# 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

| Component | Supported |
|-----------|-----------|
| **Python** | 3.9 - 3.14 (some extras require ≥3.10, see Installation) |
| **OS** | Linux, macOS, Windows |
| **Rust** | Not required for Linux (x86_64, aarch64), macOS (arm64), or Windows (x64) wheels. Other platforms build from source |

## Documentation

- **[Quickstart](https://tenuo.ai/quickstart)** - Get running in 5 minutes
- **[OpenAI](https://tenuo.ai/openai)** - Direct API protection with streaming defense
- **[Google ADK](https://tenuo.ai/google-adk)** - ADK agent tool protection
- **[AutoGen](https://tenuo.ai/autogen)** - AgentChat tool protection
- **[A2A](https://tenuo.ai/a2a)** - Inter-agent delegation with warrants
- **[FastAPI](https://tenuo.ai/fastapi)** - Zero-boilerplate API protection
- **[LangChain](https://tenuo.ai/langchain)** - Tool protection
- **[LangGraph](https://tenuo.ai/langgraph)** - Multi-agent security
- **[CrewAI](https://tenuo.ai/crewai)** - Multi-agent crew protection
- **[MCP](https://tenuo.ai/mcp)** - Model Context Protocol client + server verification
- **[Temporal](https://tenuo.ai/temporal)** - Workflow + activity authorization (replay-safe)
- **[Security](https://tenuo.ai/security)** - Threat model, best practices
- **[Tenuo Cloud](https://docs.tenuo.ai)** - Optional managed control plane (API + proprietary [`tenuo-cloud`](https://pypi.org/project/tenuo-cloud/) SDK)
- **[API Reference](https://tenuo.ai/api-reference)** - Full SDK docs

## License

Apache-2.0. See [LICENSE](../LICENSE) for details.

