> ## Documentation Index > Fetch the complete documentation index at: https://docs.nanny.run/llms.txt > Use this file to discover all available pages before exploring further. # Python > Per-function governance for Python agents using Nanny decorators. The Python SDK brings the same enforcement model as the [Rust SDK](/v0.5/guides/rust-sdk) to Python, `@tool`, `@rule`, and `@agent` decorators that enforce limits per function call. ```bash theme={null} pip install nanny-sdk ``` *** ## Passthrough mode When running outside `nanny run`, every decorator is a no-op. The function executes normally with no enforcement overhead: ```bash theme={null} # Governed — enforcement active (reads [start].cmd from nanny.toml) nanny run # Not governed — decorators silent, agent runs normally python agent.py uv run agent.py ``` *** ## `@tool` — declare a governed tool Mark a function as a tool that Nanny should track and charge against the budget: ```python theme={null} from nanny_sdk import tool @tool(tokens=10) def fetch_page(url: str) -> str: import httpx return httpx.get(url).text ``` When the agent calls `fetch_page`: 1. Nanny checks: is `fetch_page` in the `[tools] allowed` list? 2. Nanny checks: has `fetch_page` exceeded `[tools.fetch_page] max_calls`? 3. Nanny charges 10 tokens against the budget. 4. If any check fails, a `NannyStop` exception is raised — the function body never runs. Works identically for async functions: ```python theme={null} @tool(tokens=10) async def fetch_page(url: str) -> str: import httpx async with httpx.AsyncClient() as client: return (await client.get(url)).text ``` ### Tokens The `tokens` argument is required. Set it to `0` for tools you want tracked but not charged: ```python theme={null} @tool(tokens=0) def log_step(msg: str) -> None: ... ``` ### Matching the tool allowlist The tool name used for allowlist checks is the **function name** as declared in Python: ```toml theme={null} # nanny.toml [tools] allowed = ["fetch_page", "read_file"] [tools.fetch_page] max_calls = 20 tokens_per_call = 10 # nanny.toml value overrides the decorator default ``` metrics_crew — ToolDenied fires when the analysis agent calls write_report, a tool outside its allowlist *** ## `instrument` — automatic LLM token tracking Call `nanny_sdk.instrument(client)` once at agent startup to automatically report LLM token usage to Nanny's budget. Every completion response is intercepted and its token counts are debited from the same ledger that `@tool` charges against. ```python theme={null} import openai import nanny_sdk client = openai.OpenAI() nanny_sdk.instrument(client) # one line — done # From here on, every response's token usage is reported to Nanny. response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello"}], ) ``` Supported clients (detected by duck-typing — no provider package is imported): * **OpenAI, Groq, Together AI, Azure OpenAI, LiteLLM** — any client with a `chat.completions.create` method * **Anthropic** — `client.messages.create` * **Mistral** — `client.chat.complete` * **Google Gemini** (`google-genai` SDK) — `client.models.generate_content` * **Cohere v2** — `cohere.ClientV2` `instrument` returns the same client unchanged — it patches the method in-place. In passthrough mode (outside `nanny run`), it is a no-op: no wrapping, no overhead. For providers that report prompt-caching usage (OpenAI, Anthropic, DeepSeek, Gemini), `instrument` also captures it and reports it as `cache_read`/`cache_write` — a finer split of `input`, never additional tokens beyond it, and never used for enforcement (Nanny still debits `input + output` exactly as before). Every provider names and shapes this data differently, so `instrument` normalizes each one's own vocabulary into these two generic fields; absent, not zero, for a provider/response that doesn't report cache usage at all. Exists purely so a downstream cost calculator can price cache-hit tokens at their real, much cheaper rate — see [`LlmUsageRecorded`](/v0.5/reference/events#llmusagerecorded) for the full field reference. *** ## `@rule` — declare an enforcement rule A rule is a function that returns a verdict on whether execution should continue. Return `True` to allow, `False` to deny: ```python theme={null} from nanny_sdk import rule @rule("no_spiral") def check_spiral(ctx) -> bool: h = ctx.tool_call_history # Deny if the last three tool calls were all the same return not (len(h) >= 3 and len(set(h[-3:])) == 1) ``` Rules are evaluated client-side on every tool call, before Nanny’s enforcement layer is contacted. When a rule returns `False`, Nanny raises: ```python theme={null} RuleDenied("no_spiral") ``` The denied tool never runs and no tokens are charged. ### PolicyContext fields The `ctx` parameter gives you a snapshot of the current execution state: | Field | Type | Description | | ------------------- | ---------------- | ------------------------------------------ | | `step_count` | `int` | Steps completed so far | | `elapsed_ms` | `int` | Wall-clock time elapsed | | `tokens_spent` | `int` | Total tokens spent | | `tool_call_counts` | `dict[str, int]` | Per-tool call counts | | `tool_call_history` | `list[str]` | Ordered log of tool names called | | `requested_tool` | `str \| None` | The tool being evaluated right now | | `last_tool_args` | `dict[str, str]` | Arguments of the tool call being evaluated | Rules are evaluated **before** the tool runs — `requested_tool` is set to the tool name being checked. Use `last_tool_args` for content-based enforcement: ```python theme={null} @rule("no_sensitive_files") def block_sensitive(ctx) -> bool: path = ctx.last_tool_args.get("path", "") return ".env" not in path and "secret" not in path ``` All fields are live. Before each rule evaluation, the SDK fetches current counters from Nanny’s enforcement layer — `step_count`, `tokens_spent`, `tool_call_counts`, and `tool_call_history` reflect the actual execution state at the moment your rule runs. metrics_crew — RuleDenied fires on the loop-detection rule, stopping the analysis agent from repeating the same computation *** ## `@agent` — activate named limits for a scope In a multi-agent system, each agent has a different role and a different risk profile. The analysis agent makes expensive API calls and deserves a tight token ceiling. The reporter just writes a file and barely needs a budget at all. `@agent` activates the right named limit set when each role runs, then reverts automatically when it's done. ```python theme={null} from nanny_sdk import agent @agent("researcher") def run_research(topic: str) -> list[str]: # Runs under [limits.researcher] from nanny.toml pages = [fetch_page(f"https://en.wikipedia.org/wiki/{topic}")] return pages ``` ```toml theme={null} # nanny.toml [limits.researcher] steps = 200 tokens = 5000 timeout = 120000 ``` The named set inherits from `[limits]` and overrides only the declared fields. Works identically for async functions. Limits revert on exit whether the function returns normally or raises. metrics_crew running under nanny run — ingestion, analysis, visualization, and reporter agent scopes entering and exiting with structured NDJSON events *** ## `fresh_run` — starting a fresh run mid-process `@agent` changes which ceiling the run's *one* running total is checked against; it does not give a role its own budget (see [Named sets share one counter](/v0.5/concepts/limits#named-sets-share-one-counter-they-are-not-separate-budgets)). If your process genuinely runs multiple, independent phases back to back, say, a research phase that hands off to a drafting phase, or a long-lived server that should give each incoming request its own clean slate, and you want each one to start from zero rather than inherit whatever an earlier phase already spent, that's a new **run**, Nanny's real unit of governance: one cumulative counter, one stop state, "a stop is final." ```python theme={null} import nanny_sdk nanny_sdk.fresh_run() # everything governed after this point is a fresh run ``` Only meaningful when governed through a network server (`nanny run --serve` / `--join`): the server keys independent state per run, so a stop in the run you just left has zero effect on the one you're starting. Under local `nanny run` (no `--serve`), one process is already always exactly one run, so this is a safe no-op there, code that might run under either mode doesn't need to branch on which one it's in. ```python theme={null} from nanny_sdk import agent, fresh_run def handle_request(payload: dict) -> dict: fresh_run() # this request's tokens/steps start from zero, unrelated to # whatever the previous request already spent return process(payload) @agent("researcher") def research(topic: str) -> list[str]: ... ``` Before this existed, the only way to do this was setting the `NANNY_RUN_ID` environment variable directly, an internal detail the client happens to read fresh on every call, never documented as something to rely on. `fresh_run()` is that same mechanism, given a real, discoverable name. *** ## What happens on stop When Nanny stops execution, it raises a `NannyStop` exception. All stop reasons are distinct subclasses: ```python theme={null} from nanny_sdk import ( NannyStop, MaxStepsReached, BudgetExhausted, TimeoutExpired, ToolDenied, RuleDenied, AgentCompleted, AgentNotFound, BridgeUnavailable, ) ``` Catch them by category or individually: ```python theme={null} from nanny_sdk import NannyStop, BudgetExhausted, ToolDenied try: run_research("Alan Turing") except BudgetExhausted: print("Hit the token ceiling") except ToolDenied as e: print(f"Blocked tool: {e.tool_name}") except NannyStop as e: print(f"Stopped: {type(e).__name__}") ``` `BridgeUnavailable` is raised when Nanny’s enforcement layer is active but unreachable during rule evaluation or a tool call. Nanny fails closed — the agent does not continue ungoverned. Like all `NannyStop` subclasses it extends `BaseException`, so it propagates through broad `except Exception` handlers in agent frameworks without being swallowed. Under plain `nanny run` (single process, no `--serve`), you do not need to handle stop reasons in most agent code. They propagate up the call stack and terminate the process, since `nanny run` is watching that process and exits it for you. Catching them there is useful mainly in test code and at the CLI entry point. **This is not true once you're running under `nanny run --serve` / `--join`.** When your process joins a governance server, the wrapping `nanny run --join` command does not watch your process for stops at all: no polling loop, no kill on budget exhaustion, nothing. That's deliberate. [The stop guarantee](/v0.5/guides/governance-server#long-lived-processes-and-nannystop) is that a stop ends one *run*, not your whole process, so a long-lived server (a web app, a Discord bot, anything handling more than one request per process) has to keep running after another request's run gets stopped. Nothing does that for you. If a request handler lets `NannyStop` propagate uncaught, your framework's own crash behavior decides what happens next, which for many servers means the whole process going down over one governed request. Catch it at the boundary of whatever "one governed unit of work" means for your app, typically each request handler, and turn it into a failure response for that one request, not an unhandled exception: ```python theme={null} from nanny_sdk import fresh_run, NannyStop def handle_request(payload: dict) -> dict: fresh_run() # this request's budget starts clean, independent of others try: return process(payload) except NannyStop as e: return {"error": f"stopped: {type(e).__name__}"} ``` See [Long-lived processes and NannyStop](/v0.5/guides/governance-server#long-lived-processes-and-nannystop) for the full explanation of why this only applies under `--serve`/`--join`. metrics_crew — BudgetExhausted stops the analysis agent when it hits its token ceiling *** ## Complete example ```python theme={null} from nanny_sdk import tool, rule, agent @tool(tokens=10) def fetch_page(url: str) -> str: import httpx return httpx.get(url).text @tool(tokens=5) def read_file(path: str) -> str: with open(path) as f: return f.read() @rule("no_spiral") def check_spiral(ctx) -> bool: h = ctx.tool_call_history return not (len(h) >= 3 and len(set(h[-3:])) == 1) @agent("researcher") def research(topic: str) -> list[str]: results = [] page = fetch_page(f"https://en.wikipedia.org/wiki/{topic}") results.append(page) return results if __name__ == "__main__": pages = research("Alan Turing") print(f"Collected {len(pages)} pages") ``` Run it under Nanny: ```bash theme={null} nanny run ``` Run it without Nanny (decorators silent, agent runs normally): ```bash theme={null} python agent.py ``` *** ## Multi-agent pattern The canonical use case: a pipeline where each agent has a specific role, a specific budget, and access to only the tools it needs. This is the `metrics_crew` pattern — four specialized agents, each governed independently. ```python theme={null} from nanny_sdk import tool, rule, agent from collections import deque # Each tool declares its token cost. The decorator fires on every call # regardless of which agent invoked it. @tool(tokens=10) def compute_stats(metric: str, path: str) -> dict: ... @tool(tokens=10) def detect_anomalies(metric: str, path: str) -> list: ... @tool(tokens=5) def write_report(content: str, output_path: str) -> str: ... # A rule that prevents the analysis agent from looping on the same computation. _recent: deque[str] = deque(maxlen=5) @rule("no_analysis_loop") def check_loop(ctx) -> bool: tool = ctx.requested_tool or "" _recent.append(tool) return not (len(_recent) == 5 and all(t == "compute_stats" for t in _recent)) # Each agent activates its own limit scope. @agent("analysis") def run_analysis(path: str): # Governed by [limits.analysis]: steps=60, tokens=200, timeout=60000 # Tool allowlist: ["compute_stats", "detect_anomalies"] only # write_report() here would raise ToolDenied immediately stats = compute_stats("cpu_usage", path) anomalies = detect_anomalies("cpu_usage", path) return anomalies @agent("reporter") def run_reporter(findings: list, output_dir: str): # Governed by [limits.reporter]: steps=20, tokens=50, timeout=30000 # Tool allowlist: ["write_report"] only # compute_stats() here would raise ToolDenied immediately return write_report(str(findings), f"{output_dir}/report.md") ``` ```toml theme={null} # nanny.toml [limits] steps = 200 tokens = 500 timeout = 120000 [limits.analysis] steps = 60 tokens = 200 timeout = 60000 [limits.reporter] steps = 20 tokens = 50 timeout = 30000 [tools] allowed = ["compute_stats", "detect_anomalies", "write_report"] ``` The key properties this gives you: * **A ceiling tailored to each role:** the analysis agent gets a tighter check than the reporter while it's running, expressed once in `nanny.toml`. Worth being precise about what this is *not*: tokens spent is one running total for the whole run, not a separate pool per role, so if analysis exhausts its own ceiling, the run stops there, the reporter never gets to run at all. See [Named sets share one counter](/v0.5/concepts/limits#named-sets-share-one-counter-they-are-not-separate-budgets) for the full explanation, and `fresh_run()` below if you actually want each phase to have its own independent budget. * **Least-privilege tool access:** each agent only receives the tools it needs; calling outside its role raises `ToolDenied` immediately * **Loop detection:** the `@rule` fires client-side before Nanny’s enforcement layer is contacted — the denied tool never runs and no tokens are charged * **Full audit trail:** every tool call, every limit activation, every stop reason logged to NDJSON metrics_crew — MaxStepsReached stops the analysis agent after it exhausts its step allowance For cross-process and cross-machine enforcement, use the [governance server](/v0.5/guides/governance-server). See [`examples/python/metrics_crew`](https://github.com/nanny-run/nanny/tree/main/examples/python/metrics_crew) for the complete working implementation of this pattern with four agents, Plotly chart generation, and a full incident report output. *** ## Framework integration ### LangChain Stack `@lc_tool` (outer) and `@nanny_tool` (inner). LangChain registers the function for dispatch; Nanny intercepts every call regardless of which model or API style invoked it: ```python theme={null} from langchain_core.tools import tool as lc_tool from nanny_sdk import tool as nanny_tool @lc_tool # outer — LangChain registers for tool dispatch @nanny_tool(tokens=5) # inner — Nanny intercepts before file is opened def read_file(path: str) -> str: """Read a source file from disk.""" with open(path) as f: return f.read() ``` Execution order: your code calls `tool.run(args)` → LangChain validates args → Nanny wrapper intercepts → enforcement check → if allowed, file is read. ### CrewAI Same stacking pattern. CrewAI's `@tool` decorator and Nanny's `@tool` decorator both wrap the function — Nanny's wrapper fires on every `tool.run()` call inside the crew: ```python theme={null} from crewai.tools import tool as crew_tool from nanny_sdk import tool as nanny_tool @crew_tool # outer — CrewAI registers for agent dispatch @nanny_tool(tokens=15) # inner — Nanny intercepts before function runs def generate_chart(metric: str, output_dir: str) -> str: """Generate an interactive Plotly chart for a metric.""" # ... chart generation ... return output_path ``` See [`examples/python/dev_assist`](https://github.com/nanny-run/nanny/tree/main/examples/python/dev_assist) for a complete LangGraph integration and [`examples/python/metrics_crew`](https://github.com/nanny-run/nanny/tree/main/examples/python/metrics_crew) for the canonical multi-agent governance example with four specialized agents, per-role limits, and per-role tool allowlists.