Python

pip install nanny-sdk

# 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

from nanny_sdk import tool

@tool(tokens=10)
def fetch_page(url: str) -> str:
    import httpx
    return httpx.get(url).text

@tool(tokens=10)
async def fetch_page(url: str) -> str:
    import httpx
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text

@tool(tokens=0)
def log_step(msg: str) -> None: ...

# 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

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"}],
)

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)

RuleDenied("no_spiral")

@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

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

# nanny.toml
[limits.researcher]
steps   = 200
tokens  = 5000
timeout = 120000

import nanny_sdk

nanny_sdk.fresh_run()   # everything governed after this point is a fresh run

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

from nanny_sdk import (
    NannyStop,
    MaxStepsReached,
    BudgetExhausted,
    TimeoutExpired,
    ToolDenied,
    RuleDenied,
    AgentCompleted,
    AgentNotFound,
    BridgeUnavailable,
)

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__}")

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__}"}

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

nanny run

python agent.py

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

# 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"]

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

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