Tenuo Python SDK API Reference

from tenuo import MAX_DELEGATION_DEPTH, MAX_WARRANT_SIZE, MAX_WARRANT_TTL_SECS, DEFAULT_WARRANT_TTL_SECS

from tenuo import configure, SigningKey

# Development (self-signed warrants)
kp = SigningKey.generate()  # In production: SigningKey.from_env("MY_KEY")
configure(
    issuer_key=kp,
    dev_mode=True,
    allow_self_signed=True,
)

# Production (trusted roots required)
configure(
    issuer_key=my_keypair,
    trusted_roots=[control_plane_pubkey],
)

from tenuo import auto_configure

# Reads TENUO_* environment variables automatically
auto_configure()

from tenuo import get_config

config = get_config()
print(f"TTL: {config.default_ttl}")
print(f"Dev mode: {config.dev_mode}")
print(f"Mode: {config.mode}")  # EnforcementMode enum

from tenuo import EnforcementMode, is_audit_mode, is_enforce_mode, should_block_violation

# Check current mode
if is_audit_mode():
    print("Violations are logged but not blocked")

if should_block_violation():
    raise AuthorizationDenied(...)

from tenuo import SigningKey

from tenuo import PublicKey

from tenuo import Signature

from tenuo import Warrant

from tenuo import Warrant, Pattern

# Fluent builder pattern
warrant = (Warrant.mint_builder()
    .capability("read_file", path=Pattern("/data/*"))
    .holder(worker_key.public_key)
    .ttl(3600)
    .mint(issuer_key))

# Delegate with narrower scope
child = (parent.grant_builder()
    .capability("read_file", path=Pattern("/data/reports/*"))
    .holder(worker_key.public_key)
    .ttl(300)
    .grant(parent_key))  # Parent holder signs

from tenuo import Warrant, Pattern, Range, Clearance

# Execution warrant with builder
warrant = (Warrant.mint_builder()
    .capability("read_file",
        path=Pattern("/data/*"),
        max_size=Range(max=1000000))
    .ttl(3600)
    .holder(keypair.public_key)
    .mint(keypair))

# Issuer warrant with builder
issuer = (Warrant.mint_builder()
    .issuer()  # Switch to issuer mode
    .issuable_tools(["read_file", "write_file"])
    .clearance(Clearance.INTERNAL)  # Optional
    .constraint_bound("path", Pattern("/data/*"))
    .max_issue_depth(3)
    .mint(keypair))

# Property examples
warrant.ttl_remaining       # timedelta(seconds=299, ...)
warrant.expires_at          # datetime(2025, 12, 22, 21, 30, ...)
warrant.is_expired          # False
warrant.is_terminal         # False  
warrant.capabilities        # {'tools': ['read_file'], 'path': '/data/*', 'max_size': 1000000}

from tenuo import Warrant, WhyDenied, DenyCode

# Logic Check (UX-only, no crypto)
# "Does the warrant allow this?"
if warrant.allows("read_file", args={"path": "/data/report.txt"}):
    print("Allowed by logic")
else:
    print(f"Would be denied")

# Why denied (for debugging)
reason = warrant.why_denied("write_file", path="/etc/passwd")
# WhyDenied(deny_code=DenyCode.TOOL_NOT_ALLOWED, tool='write_file', ...)

if reason.deny_code == DenyCode.TOOL_NOT_ALLOWED:
    print("Tool not in warrant")

# Generate HTTP headers
headers = warrant.headers(keypair, "read_file", {"path": "/data/x.txt"})
# {'X-Tenuo-Warrant': '<base64>', 'X-Tenuo-PoP': '<signature>'}

print(repr(warrant))
# <Warrant id=tnu_wrt_019b... tools=[read_file, write_file] ttl=0:04:59>

# Many tools are truncated
# <Warrant id=tnu_wrt_019b... tools=[a, b, c, +3 more] ttl=0:04:59>

# Child only gets what you explicitly grant
builder = parent.grant_builder()
builder.capability("read_file", path=Exact("/data/report.txt"))
builder.holder(worker_key.public_key)
child = builder.grant(parent_key)
# child.tools == ["read_file"] (only!)

# Start with all parent capabilities, then narrow
builder = parent.grant_builder()
builder.inherit_all()                    # Explicit opt-in
builder.tools(["read_file"])             # Keep only this tool
builder.holder(worker_key.public_key)
child = builder.grant(parent_key)

with key_scope(my_keypair):
    child = parent.grant(
        holder=worker.public_key,
        tools=["read_file"],  # Narrow tools
        path=Exact("/data/q3.pdf"),  # Narrow constraints
    )

# Create parent warrant, then delegate with specific tools
parent_warrant = (Warrant.mint_builder()
    .tool("read_file")
    .tool("send_email")
    .holder(control_plane_key.public_key)
    .ttl(3600)
    .mint(control_plane_key))

# Delegate to worker with narrower scope
builder = parent_warrant.grant_builder()
builder.tool("read_file")  # Only read, not send_email
builder.holder(worker_key.public_key)
builder.ttl(300)
exec_warrant = builder.grant(control_plane_key)

# Create terminal warrant (cannot be delegated further)
builder = parent.grant_builder()
builder.inherit_all()     # Inherit parent capabilities
builder.terminal()        # Mark as terminal
builder.holder(worker_key.public_key)
terminal = builder.grant(parent_key)

assert terminal.is_terminal()  # True
# terminal.grant_builder().grant(...) will fail

from tenuo import Warrant, BoundWarrant

# Create bound warrant
warrant, keypair = Warrant.quick_mint(["read_file"], ttl=300)
bound = BoundWarrant(warrant, keypair)

# Or bind from existing warrant
bound = warrant.bind(keypair)

# Delegation with bound key (no keypair arg needed)
child = bound.grant(to=worker.public_key, allow="read_file", ttl=300)

# Generate headers
headers = bound.headers("read_file", {"path": "/data/x.txt"})

# Serialization blocked
import pickle
pickle.dumps(bound)  # Raises TypeError!

print(repr(bound))
# <BoundWarrant id=tnu_wrt_019b... KEY_BOUND=True>

builder = parent_warrant.grant_builder()

builder.holder()       # Returns configured holder or None
builder.ttl()          # Returns configured TTL or None
builder.clearance()    # Returns configured clearance level or None
builder.intent()       # Returns configured intent or None

builder = warrant.grant_builder()

# Pattern 1: Grant specific capability (POLA default)
child = (parent.grant_builder()
    .capability("read_file", path=Exact("/data/q3.pdf"))
    .holder(worker_key.public_key)
    .grant(parent_key))

# Pattern 2: Inherit all, then narrow
child = (parent.grant_builder()
    .inherit_all()                    # Explicit opt-in
    .tools(["read_file"])             # Keep only this tool
    .holder(worker_key.public_key)
    .grant(parent_key))

# Reading builder state
assert builder.holder() == worker_key.public_key
assert builder.ttl() == 300

from tenuo import Authorizer

Authorizer(
    trusted_roots: Optional[List[PublicKey]] = None,
    clock_tolerance_secs: int = 30,
    pop_window_secs: int = 30,
    pop_max_windows: int = 4,
)

from tenuo import Authorizer, Clearance

authorizer = Authorizer(trusted_roots=[root_key])

# Require specific clearance levels for tools
authorizer.require_clearance("*", Clearance.EXTERNAL)        # Default baseline
authorizer.require_clearance("delete_*", Clearance.PRIVILEGED)  # Prefix pattern
authorizer.require_clearance("admin_reset", Clearance.SYSTEM)   # Exact match

# Check what's required for a tool
print(authorizer.get_required_clearance("delete_file"))  # Clearance.PRIVILEGED

from tenuo import (
    Pattern,    # Glob patterns: "staging-*"
    Exact,      # Exact match: "production"
    Range,      # Numeric ranges: Range(min=0, max=100)
    OneOf,      # Allowlist: OneOf(["read", "write"])
    NotOneOf,   # Denylist: NotOneOf(["admin"])
    Regex,      # Regular expressions: Regex("^[a-z]+$")
    Wildcard,   # Match anything (use sparingly)
    Contains,   # List contains: Contains(["admin"])
    Subset,     # List subset: Subset(["read", "write"])
    All,        # AND logic: All([constraint1, constraint2])
    AnyOf,      # OR logic: AnyOf([constraint1, constraint2])
    Not,        # Negation: Not(constraint)
    CEL,        # CEL expressions: CEL('value > 0')
)

Pattern("staging-*")      # Matches staging-web, staging-db
Pattern("/tmp/**")        # Recursive: matches /tmp/foo/bar/file.txt
Pattern("*-safe")         # Suffix: matches image-safe

Exact("production")       # Only matches "production"

Range(min=0, max=100)     # 0 <= value <= 100
Range.min_value(10)       # value >= 10
Range.max_value(1000)     # value <= 1000

OneOf(["read", "write", "delete"])   # Must be one of these
NotOneOf(["admin", "root"])          # Anything except these

Regex("^[a-z0-9_]+$")     # Matches lowercase alphanumeric with underscores
Regex(".*\\.pdf$")        # Matches files ending in .pdf

Wildcard()                # Matches anything

Contains(["admin"])                   # List must include "admin"
Subset(["read", "write", "admin"])   # Only these values allowed

All([Range.min_value(0), Range.max_value(100)])   # AND
AnyOf([Exact("admin"), Exact("superuser")])       # OR
Not(Exact("blocked"))                              # Negation

from tenuo import CEL

# Simple comparison
CEL('amount < 10000 && amount > 0')

# Multi-parameter logic
CEL('budget < revenue * 0.1 && currency == "USD"')

# With standard library functions
CEL('time_since(created_at) < 3600')  # Within last hour
CEL('net_in_cidr(ip, "10.0.0.0/8")')  # From private network

from tenuo.templates import FileReader, FileWriter, DatabaseReader, WebSearcher, CommonAgents

from tenuo import mint
from tenuo.templates import FileReader, FileWriter

# Read-only access to a directory
async with mint(FileReader.in_directory("/data/reports")) as w:
    content = read_file("/data/reports/q4.txt")  # ✓ allowed
    content = read_file("/etc/passwd")  # ✗ denied

# Read a specific file only
async with mint(FileReader.exact_file("/config/app.json")) as w:
    content = read_file("/config/app.json")  # ✓ allowed

# Read files with specific extensions
async with mint(FileReader.extensions("/docs", [".md", ".txt"])) as w:
    read_file("/docs/readme.md")  # ✓ allowed
    read_file("/docs/data.json")  # ✗ denied

# Write access (use with caution)
async with mint(FileWriter.in_directory("/tmp/agent-output")) as w:
    write_file("/tmp/agent-output/report.txt", data)  # ✓ allowed

from tenuo.templates import DatabaseReader, DatabaseWriter

# Read from specific tables
async with mint(DatabaseReader.tables(["users", "products"])) as w:
    query("SELECT * FROM users")  # ✓ allowed
    query("SELECT * FROM transactions")  # ✗ denied

# Read with row limit (prevent data exfiltration)
async with mint(DatabaseReader.with_row_limit(["users"], max_rows=10)) as w:
    query("SELECT * FROM users LIMIT 10")  # ✓ allowed

# Full-table access within a schema
async with mint(DatabaseReader.schema("public")) as w:
    query("SELECT * FROM public.users")  # ✓ allowed

from tenuo.templates import WebSearcher, ApiClient

# Web search with domain restrictions
async with mint(WebSearcher.domains(["api.openai.com", "*.google.com"])) as w:
    search("openai docs", domain="api.openai.com")  # ✓ allowed

# API client with method restrictions
async with mint(ApiClient.readonly("api.example.com")) as w:
    get("/users")  # ✓ allowed
    post("/users", {...})  # ✗ denied

from tenuo.templates import CommonAgents

# Research agent: read-only web search + file reading
async with mint(*CommonAgents.research_agent("/data/docs")) as w:
    ...

# Writer agent: file writing to specific directory
async with mint(*CommonAgents.writer_agent("/output")) as w:
    ...

# Analyst agent: database read + search
async with mint(*CommonAgents.analyst_agent(tables=["metrics", "reports"])) as w:
    ...

from tenuo import mint, Capability, Pattern

async with mint(Capability("read_file", path=Pattern("/data/*"))) as warrant:
    result = await agent.invoke(prompt)

from tenuo import mint_sync

with mint_sync(Capability("read_file", path="/data/*")) as warrant:
    result = protected_read_file(path="/data/report.csv")

from tenuo import grant

async with mint(
    Capability("read_file", path="/data/*"), 
    Capability("write_file", path="/data/*")
):
    async with grant(Capability("read_file", path="/data/reports/*")):
        # Narrower scope here
        result = await agent.invoke(prompt)

scope = grant(Capability("read_file", path="/data/reports/*"))
scope.preview().print()  # See diff before entering
async with scope:
    ...

from tenuo import guard_tools

guard_tools(
    tools: List[Any],
    *,
    inplace: bool = True,
    strict: bool = False,
    schemas: Optional[Dict[str, ToolSchema]] = None,
) -> List[Any]

from tenuo import guard_tools, mint

# Define your tools
tools = [read_file, send_email, query_db]

# Wrap them (mutates in place by default)
guard_tools(tools)

# Use with scoped authority
async with mint(Capability("read_file", path="/data/*")):
    result = await tools[0](path="/data/report.csv")

original = [read_file, send_email]
protected = guard_tools(original, inplace=False)
# original unchanged, protected has wrapped tools

from tenuo import McpConfig, CompiledMcpConfig

config = McpConfig.from_file("mcp-config.yaml")

compiled = CompiledMcpConfig.compile(config)
result = compiled.extract_constraints("filesystem_read", {"path": "/var/log/app.log"})

    from tenuo import guard

@guard(
    warrant_or_tool=None,  # Warrant instance OR tool name string
    tool=None,             # Tool name (if not passed as first arg)
    keypair=None,          # SigningKey for PoP (or use context)
    extract_args=None,     # Optional custom extractor function
    mapping=None,          # Arg name → constraint name mapping
)

@guard(tool="query")
def query_db(query: str, table: str = "users", limit: int = 100):
    ...

# All arguments extracted automatically:
query_db("SELECT *")  
# → {query: "SELECT *", table: "users", limit: 100}
#                        ↑ defaults included

@guard(
    tool="transfer",
    extract_args=lambda from_account, to_account, amount, **kw: {
        "source": from_account,
        "destination": to_account,
        "amount": amount
    }
)
def transfer(from_account: str, to_account: str, amount: float):
    ...

@guard(
    tool="read_file",
    mapping={"file_path": "path"}  # Rename after extraction
)
def read_file(file_path: str):
    ...
# Extracted as: {path: "..."}

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

# Use with task scoping
async with mint(Capability("read_file", path=Pattern("/data/*"))):
    read_file("/data/test.txt")

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

@guard("read_file")  # Shorthand: tool name as positional arg
def read_file(path: str) -> str:
    return open(path).read()

from tenuo import (
    warrant_scope,
    get_warrant_context,
    key_scope,
    get_signing_key_context,
)

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

# Create warrant and bind key
keypair = SigningKey.generate()  # In production: SigningKey.from_env("MY_KEY")
warrant = Warrant.mint_builder().tool("search").tool("calculator").mint(keypair)
bound = warrant.bind(keypair)

# Protect tools
protected_tools = guard([search, calculator], bound)

# Use in your agent
agent = create_openai_tools_agent(llm, protected_tools, prompt)

from tenuo.langgraph import tenuo_node
from tenuo import Capability, Pattern

@tenuo_node
async def researcher(state, bound_warrant):
    # bound_warrant is injected automatically
    if bound_warrant.allows("search"):
        results = await search_tool(query=state["query"])
        return {"results": results}
    return {"messages": ["Not authorized"]}

from tenuo.langgraph import require_warrant

@require_warrant
async def sensitive_node(state):
    ...

from tenuo.langgraph import TenuoToolNode
from tenuo import mint_sync

# Before (manual protection):
# tools = [search, calculator]
# protected = protect_langchain_tools(tools)
# tool_node = ToolNode(protected)

# After (automatic protection):
tool_node = TenuoToolNode([search, calculator])

graph.add_node("tools", tool_node)

# Run with authorization
with mint_sync(Capability("search"), Capability("calculator")):
    result = graph.invoke(...)

from tenuo import KeyRegistry, SigningKey

registry = KeyRegistry.get_instance()

# Register keys
registry.register("worker", SigningKey.from_env("WORKER_KEY"))
registry.register("orchestrator", SigningKey.from_env("ORCH_KEY"))

# Retrieve
key = registry.get("worker")

# Multi-tenant: namespace keys per tenant
registry.register("api", tenant_key, namespace="tenant-123")
key = registry.get("api", namespace="tenant-123")

 from tenuo.fastapi import TenuoGuard, SecurityContext, require_warrant

 from fastapi import FastAPI
 from tenuo import configure, SigningKey
 from tenuo.fastapi import TenuoGuard
 
 app = FastAPI()
 
 app.add_middleware(
     TenuoGuard,
     # Optional config overrides
     # trusted_roots=[...], 
     # verbose_errors=True
 )

 @app.get("/secure")
 async def secure_endpoint(
     ctx: SecurityContext = Depends(require_warrant)
 ):
     return {"warrant_id": ctx.warrant_id}

from tenuo.testing import (
    allow_all,
    assert_authorized,
    assert_denied,
    assert_can_grant,
    assert_cannot_grant,
    deterministic_headers,
)

from tenuo import guard
from tenuo.testing import allow_all

@guard(tool="dangerous_action")
def dangerous_action():
    return "executed"

# In tests (pytest, unittest, or TENUO_TEST_MODE=1)
def test_dangerous_action():
    with allow_all():
        result = dangerous_action()  # No warrant needed!
        assert result == "executed"

from tenuo import guard
from tenuo.testing import assert_authorized, assert_denied

@guard(tool="read_file")
def read_file(path: str):
    return f"Content of {path}"

def test_authorization():
    # Assert code succeeds
    with assert_authorized():
        read_file("/data/report.txt")
    
    # Assert code is denied (with optional code/reason check)
    with assert_denied(code="ConstraintViolation"):
        read_file("/etc/passwd")
    
    # Assert with custom message
    with assert_denied(message="Should block access to system files"):
        read_file("/etc/shadow")

from tenuo import Warrant
from tenuo.testing import assert_can_grant, assert_cannot_grant

def test_delegation_chain():
    # Create a root warrant
    root, root_key = Warrant.quick_mint(["search", "read_file"], ttl=3600)
    
    # Assert we CAN grant a subset of tools
    child, child_key = assert_can_grant(
        root, root_key,
        child_tools=["read_file"],  # Subset of parent
    )
    
    # Assert we CANNOT grant tools not in parent
    assert_cannot_grant(
        root, root_key,
        child_tools=["delete_file"],  # Not in parent!
        expected_reason="ToolNotAuthorized",
    )

from tenuo import Warrant

# Returns (warrant, signing_key)
warrant, key = Warrant.quick_mint(["read_file", "search"], ttl=300)

# Use the warrant
bound = warrant.bind(key)
headers = bound.headers("search", {"query": "test"})

from tenuo import Warrant

def test_my_function():
    warrant = Warrant.for_testing(["read_file", "write_file"])
    # Use for testing without real key management

from tenuo.testing import deterministic_headers

headers = deterministic_headers(warrant, key, "read_file", {"path": "/data/x"})
# Headers are deterministic for the same inputs

from tenuo import Capability, Pattern, OneOf, Exact

capabilities = [
    Capability("search", query=Pattern("*")),
    Capability("read_file", path=OneOf(["/data/reports/*", "/data/docs/*"])),
    Capability("query", table=OneOf(["users", "products"]), operation=Exact("SELECT")),
]

from tenuo import TenuoError, ScopeViolation, WarrantViolation

from tenuo import AuthorizationDenied, ConstraintResult, Pattern

# Example error output:
# Access denied for tool 'read_file'
#
#   ❌ path:
#      Expected: Pattern("/data/*")
#      Received: '/etc/passwd'
#      Reason: Pattern does not match
#   ✅ size: OK

# Create from constraint check
error = AuthorizationDenied.from_constraint_check(
    tool="read_file",
    constraints={"path": Pattern("/data/*"), "size": Range(max=1000)},
    args={"path": "/etc/passwd", "size": 500},
    failed_field="path",
    failed_reason="Pattern does not match",
)

print(error)  # Shows detailed diff

from tenuo import audit_logger, AuditEventType

audit_logger.configure(service_name="payment-service")

audit_logger.log_authorization_success(
    warrant_id="wrt_123",
    tool="process_payment",
    constraints={"amount": 100.0}
)

from tenuo import ReadableWarrant, SignableWarrant, AnyWarrant

from typing import Protocol

class ReadableWarrant(Protocol):
    @property
    def id(self) -> str: ...
    @property
    def tools(self) -> list[str]: ...
    @property
    def ttl_remaining(self) -> timedelta: ...
    @property
    def is_expired(self) -> bool: ...
    @property
    def is_terminal(self) -> bool: ...

class SignableWarrant(Protocol):
    def delegate(self, holder: PublicKey, ...) -> "Warrant": ...
    def headers(self, tool: str, args: dict) -> dict: ...

AnyWarrant = Union[Warrant, BoundWarrant]

def process_warrant(w: AnyWarrant) -> None:
    print(w.id, w.tools)  # Works for both types