> ## 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.6/guides/rust-sdk) to Python, `@tool`, `@rule`, and `@agent` decorators that govern each call before it executes. ```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 --serve # 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 govern: ```python theme={null} from nanny_sdk import tool @tool() 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 records the call, with the labels the operator declared for it. 4. If any check fails, a `NannyStop` exception is raised, the function body never runs. Works identically for async functions: ```python theme={null} @tool() async def fetch_page(url: str) -> str: import httpx async with httpx.AsyncClient() as client: return (await client.get(url)).text ``` ### 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 max_calls = 10 # most calls to this tool in one run ``` *** ## `instrument`: automatic LLM token tracking Call `nanny_sdk.instrument(client)` once at startup to report LLM token usage automatically. Every completion response is intercepted and its counts recorded. Measurement only. Nothing here stops a run: tokens are recorded so you can answer what a run cost, and Nanny never decides from them. ```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.6/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. ### PolicyContext fields The `ctx` parameter gives you a snapshot of the current execution state: | Field | Type | Description | | ------------------- | ---------------------- | ----------------------------------------------------------- | | `requested_tool` | `str \| None` | The tool being evaluated right now | | `last_tool_args` | `dict[str, str]` | Arguments of that call | | `tool_labels` | `dict[str, list[str]]` | Labels for **every** allowed tool, not only the pending one | | `tool_call_history` | `list[str]` | Ordered log of tool names already called | | `tool_call_counts` | `dict[str, int]` | Per-tool call counts | | `tokens_spent` | `int` | Tokens measured so far | | `elapsed_ms` | `int` | Wall-clock time since the run started | | `now_ms` | `int` | Wall-clock at evaluation, milliseconds since the epoch | 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: the SDK fetches current state before every rule evaluation. `tool_labels` covers every allowed tool, not just the pending one, because a rule usually needs to ask what an *already-called* tool was: ```python theme={null} @rule("no_send_after_read") def taint(ctx) -> bool: pending = ctx.requested_tool if pending is None or not ctx.tool_has(pending, "external_effect"): return True return not any(ctx.tool_has(t, "reads_untrusted") for t in ctx.tool_call_history) ``` `ctx.tool_has(tool, label)` returns `False` for an unknown tool and for an unknown label, which is the safe answer in both directions. `ctx.tools_with(label)` returns every allowed tool carrying a label, sorted. `now_ms` is supplied rather than read from the clock, so a rule about when an action is permitted stays a pure function of its inputs and can be tested. *** ## `@agent`: name a phase of the run In a multi-agent system each phase does different work, and a denial in one is worth attributing to that phase rather than to the process as a whole. `@agent` names the phase, and every event between entering and exiting belongs to it. ```python theme={null} from nanny_sdk import agent @agent("researcher") def run_research(topic: str) -> list[str]: ... ``` `AgentScopeEntered` and `AgentScopeExited` bracket the events in between, so the audit log can answer which phase a denial came from. Nesting is supported. Works identically for async functions. A scope does not change what the agent may do; it labels which phase each verdict belongs to. The scope exits whether the function returns normally or raises. *** ## `run_scope`: an independent run inside one process A **run** is Nanny's unit of governance: one rule set, one stop state, final once stopped. A long-lived server wants each request to be its own run, so a stop in one never affects another. ```python theme={null} from nanny_sdk import run_scope def handle_request(payload: dict) -> dict: with run_scope(): return process(payload) ``` Safe under concurrency: the run id lives in a context variable, so threads and async tasks each get their own rather than sharing one process-global value. The scope self-cleans on exit. Only meaningful when governed through a governance server (`nanny run --serve` or `--join`), which keys state per run. Under local `nanny run` one process is always exactly one run, so this is a no-op and code that might run under either mode does not need to branch. *** ## `get_run_events`: read a run's events from your own code Returns the events a governance server has buffered for one run: ```python theme={null} from nanny_sdk import get_run_events for event in get_run_events(run_id): if event["event"] == "RuleDenied": notify(event["rule_name"], event["tool"]) ``` You pass the `run_id` explicitly rather than it being read from the current scope, because the caller is usually a background worker polling several runs and has no run of its own. Every call returns the run's full buffered list, so track how many you have already consumed, by index or by `seq`. Returns `[]` for a run with no events yet and for an unreachable server: this is a side channel, and it should never crash a session over a blip that governed calls already handle by failing closed. *** ## `set_app`: declare which app this process is ```python theme={null} import json from pathlib import Path from nanny_sdk import set_app identity = json.loads(Path(".nanny/app.json").read_text()) set_app(identity["app_id"], identity.get("name", "")) ``` Emits `AppIdentified`, so runs from several apps sharing one governance server are attributed separately. A process that declares nothing inherits the governor's identity. `nanny run` already does this for the process it launches, under both `--serve` and `--join`. Call it yourself only for a process `nanny run` did not start, such as a worker that joins a governor from its own entrypoint. *** ## `set_harness`: declare what is running this agent ```python theme={null} from nanny_sdk import set_harness set_harness("my-agent-engine") # or set_harness("crewai", "0.5.1") ``` `instrument()` detects the well-known frameworks from the call stack and the imported modules, and reports what it found alongside each LLM call. Call this when detection cannot help, which is not a rare case: an application whose agent loop is its own code matches no framework, so detection correctly reports nothing and every run arrives unattributed. A first-party agent is not an unknown harness, it simply is not a framework. An explicit declaration wins over detection for the life of the process, since a framework being importable does not mean it drove the call. Deduped bridge-side, so declaring on every request is safe, and a no-op in passthrough mode like the decorators. *** ## 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, ToolDenied, RuleDenied, AgentCompleted, BridgeUnavailable, ExecutionStopped, ) ``` Catch them by category or individually: ```python theme={null} from nanny_sdk import NannyStop, RuleDenied, ToolDenied try: run_research("Alan Turing") except RuleDenied as e: print(f"Rule refused: {e.rule_name}") 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. A process that has **never** reached the bridge retries for 30 seconds first, with backoff, so a container started before the governor it joins waits rather than failing the work it was handed. That retry is armed once and disarms permanently on the first success: after a governor has answered, a later failure raises immediately, because retrying then would let the agent keep calling tools while nothing was authorising them. 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 a stop, nothing. That's deliberate. [The stop guarantee](/v0.6/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 run_scope, NannyStop def handle_request(payload: dict) -> dict: try: with run_scope(): return process(payload) except NannyStop as e: return {"error": f"stopped: {type(e).__name__}"} ``` See [Long-lived processes and NannyStop](/v0.6/guides/governance-server#long-lived-processes-and-nannystop) for the full explanation of why this only applies under `--serve`/`--join`. *** ## Complete example ```python theme={null} from nanny_sdk import tool, rule, agent @tool() def fetch_page(url: str) -> str: import httpx return httpx.get(url).text @tool() 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 --serve ``` Run it without Nanny (decorators silent, agent runs normally): ```bash theme={null} python agent.py ``` *** ## Multi-agent pattern A pipeline where each phase has a role, its own tools, and rules that apply across all of them. ```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() def compute_stats(metric: str, path: str) -> dict: ... @tool() def detect_anomalies(metric: str, path: str) -> list: ... @tool() 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)) # @agent names the phase, so every verdict is attributed to it. @agent("analysis") def run_analysis(path: str): stats = compute_stats("cpu_usage", path) anomalies = detect_anomalies("cpu_usage", path) return anomalies @agent("reporter") def run_reporter(findings: list, output_dir: str): return write_report(str(findings), f"{output_dir}/report.md") ``` ```toml theme={null} # nanny.toml [tools] allowed = ["compute_stats", "detect_anomalies", "write_report"] [tools.write_report] external_effect = true max_calls = 3 ``` The key properties this gives you: * **Least privilege:** a tool outside `[tools] allowed` raises `ToolDenied` before any rule runs * **Loop detection:** the `@rule` fires before Nanny's enforcement layer is contacted, so the denied tool never runs * **Attribution:** every verdict is bracketed by the `@agent` scope that produced it * **Full audit trail:** every tool call and every stop reason logged to NDJSON For cross-process and cross-machine enforcement, use the [governance server](/v0.6/guides/governance-server). *** ## 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() # 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() # 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 ``` Rule packs ship the same shape as the rules above. See [Rule packs](/v0.6/guides/rule-packs).