[AAASM-3537] ✨ (examples): Add Agno tool-policy example (re-land on master)

.github/workflows/verify-python.yml

@@ -229,6 +229,25 @@ jobs:
       - name: Run smoke tests
         run: uv run pytest tests/ -v
 
+  agno-tool-policy:
+    name: agno-tool-policy
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: python/agno-tool-policy
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
+
+      - uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39  # v8.2.0
+        with:
+          version: "latest"
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Run smoke tests
+        run: uv run pytest tests/ -v
+
   haystack-tool-policy:
     name: haystack-tool-policy
     runs-on: ubuntu-latest

README.md

@@ -81,6 +81,7 @@ below is the shared cross-language view.
 | [`python/google-adk/`](./python/google-adk/) | Google ADK | Scripted offline tool trajectory governing `BaseTool.run_async` (no cloud creds) |
 | [`python/haystack-tool-policy/`](./python/haystack-tool-policy/) | Haystack | Govern a real Haystack agent via the native adapter — real `Tool.invoke` allow/deny through a `ToolInvoker` |
 | [`python/smolagents-tool-policy/`](./python/smolagents-tool-policy/) | Smolagents | Govern real `smolagents.Tool` calls via `Tool.__call__`, blocking a destructive tool offline (no model creds) |
+| [`python/agno-tool-policy/`](./python/agno-tool-policy/) | Agno | Govern real Agno `FunctionCall.execute` tool calls (allow / deny / pending); offline, no model creds |
 
 ### Node.js / TypeScript
 

python/README.md

@@ -18,6 +18,7 @@ This directory contains runnable Python examples showing how to integrate Agent
 | `haystack-tool-policy/`            | Haystack         | Govern a real Haystack agent via the native adapter — real `Tool.invoke` allow/deny through a `ToolInvoker` |
 | `smolagents-tool-policy/`          | Smolagents       | Govern real `smolagents.Tool` calls via `Tool.__call__`, blocking a destructive tool offline (no model creds) |
 | `microsoft-agent-framework-tool-policy/` | Microsoft Agent Framework | Govern `FunctionTool.invoke` (allow / deny / pending); mock + live paths |
+| `agno-tool-policy/`                | Agno             | Govern real Agno `FunctionCall.execute` tool calls (allow / deny / pending); offline, no model creds |
 
 All examples use the `agent-assembly` Python package (available on PyPI).
 

python/agno-tool-policy/.env.example

@@ -0,0 +1,7 @@
+# Agent Assembly gateway connection (required for production mode)
+# Leave unset to run in offline / sdk-only demo mode
+# AGENT_ASSEMBLY_GATEWAY_URL=http://localhost:8080
+# AGENT_ASSEMBLY_API_KEY=your-api-key-here
+
+# Agno LLM provider (optional — this example drives tools directly, no live LLM)
+# OPENAI_API_KEY=your-openai-key-here

python/agno-tool-policy/README.md

@@ -0,0 +1,109 @@
+# agno-tool-policy
+
+Demonstrates [Agent Assembly](https://github.com/AI-agent-assembly/agent-assembly-examples) governance of an [Agno](https://docs.agno.com/) (formerly Phidata) agent using Agno's **native Agent Assembly adapter**.
+
+Agno runs every function-tool call through a single chokepoint — `agno.tools.function.FunctionCall.execute`. The SDK's Agno adapter patches that method, so every tool an Agno agent invokes passes through policy **before** its body executes: a `deny` short-circuits the tool entirely (its body never runs), an `allow` runs it. This example wires the real adapter (`AgnoPatch`) to a local offline policy and drives genuine Agno `@tool` functions exactly as an Agno `Agent` does, so you can watch the safe tools run and the dangerous one get blocked — with no gateway, API key, or live LLM.
+
+## What this example demonstrates
+
+- Initializing Agent Assembly with `init_assembly()`.
+- Governing real Agno `@tool` functions through the **native Agno adapter** (`AgnoPatch` over `FunctionCall.execute`).
+- An **allowed** tool call (`get_weather`) that runs and returns its real output.
+- Another **allowed** tool call (`summarize_docs`).
+- A **denied** tool call (`execute_sql`) whose body is short-circuited by `deny_arbitrary_execution` — the tool never executes.
+
+## Prerequisites
+
+| Requirement | Version |
+|---|---|
+| Python | >= 3.12 |
+| [uv](https://github.com/astral-sh/uv) | latest |
+| Agent Assembly Python SDK | >= 0.0.1b4 |
+| Agno | >= 2.0.0 |
+
+No API key or running gateway is required for the offline demo.
+
+## Setup
+
+```bash
+cd python/agno-tool-policy
+uv sync --extra dev
+```
+
+## Run
+
+```bash
+uv run python src/main.py
+```
+
+### Expected output
+
+```
+==============================================================
+  Agent Assembly — Agno Tool Policy Demo
+==============================================================
+
+Initializing Agent Assembly (gateway: http://localhost:8080, sdk-only mode)...
+  Agent:    agno-demo-agent
+  Gateway:  http://localhost:8080
+  Mode:     sdk-only (offline demo)
+
+Policy rules (local simulation of gateway policy):
+  DENY   — execute_sql, run_shell_command  (arbitrary execution)
+  ALLOW  — everything else
+
+Agno governance hook installed on FunctionCall.execute.
+Tools governed: get_weather, summarize_docs, execute_sql
+
+Running governed tool calls:
+--------------------------------------------
+  → get_weather({'city': 'London'})
+     ✅ ALLOWED  — Weather in London: sunny, 22C (mock)
+
+  → summarize_docs({'topic': 'policy enforcement'})
+     ✅ ALLOWED  — Summary for 'policy enforcement': Agent Assembly provides governance... (mock)
+
+  → execute_sql({'sql': 'DROP TABLE users; --'})
+     ❌ BLOCKED  — [BLOCKED by governance policy] Tool 'execute_sql' is blocked by policy rule 'deny_arbitrary_execution'. ...
+
+Assembly context shut down.
+```
+
+## Run tests
+
+```bash
+uv run pytest tests/ -v
+```
+
+The smoke tests are a real governance proof: the deny test asserts the denied tool's body **never ran** (a no-op patch would let it execute), and the allow test asserts the real tool body ran and returned its output.
+
+## How governance attaches
+
+In production, `init_assembly()` **auto-detects** Agno and installs the governance hook automatically — you do not call `AgnoPatch` yourself, and the live runtime answers each allow/deny decision. This offline demo has no live runtime, so `init_assembly()` installs a no-op hook; the example reverts it and re-applies the hook wired to a local `LocalPolicyEngine` so the allow/deny decisions are visible without a gateway.
+
+## Switching to production mode
+
+1. Start an Agent Assembly gateway or use your SaaS workspace URL.
+2. Copy `.env.example` to `.env` and fill in credentials.
+3. Drop the manual `AgnoPatch` wiring — `init_assembly()` auto-detects Agno and the live runtime governs every `FunctionCall.execute` automatically:
+
+```python
+with init_assembly(gateway_url="http://localhost:8080", agent_id="my-agno-agent") as ctx:
+    # Agno is auto-detected and governed; just run your agent as usual.
+    agent.run("...")
+```
+
+## Troubleshooting
+
+| Problem | Fix |
+|---|---|
+| `ModuleNotFoundError: agent_assembly` | Run `uv sync` first |
+| `ModuleNotFoundError: agno` | Run `uv sync` — `agno` is a required dependency |
+| `execute_sql` shows ALLOWED | The `init_assembly()` no-op hook was not reverted before applying the policy hook — the patch is idempotent, so revert it first (the example does this) |
+
+## Links
+
+- [Agent Assembly Python SDK](https://github.com/AI-agent-assembly/python-sdk)
+- [Agno docs](https://docs.agno.com/)
+- [Agent Assembly Examples](../../README.md)
+- [Python Examples](../README.md)

python/agno-tool-policy/pyproject.toml

@@ -0,0 +1,26 @@
+[project]
+name = "agno-tool-policy"
+version = "0.1.0"
+description = "Agent Assembly governance demo with the native Agno adapter"
+requires-python = ">=3.12"
+dependencies = [
+    # Requires the agent-assembly release that ships the native Agno adapter
+    # (agent_assembly.adapters.agno, AAASM-3537). Until that release is on PyPI,
+    # install the SDK from the python-sdk repo's AAASM-3537 branch.
+    "agent-assembly>=0.0.1b4",
+    "agno>=2.0.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+
+[tool.uv.sources]
+# Track the SDK's git master so the example validates against the merged
+# adapter until 0.0.1b5 publishes it to PyPI. Revert to the PyPI pin then.
+agent-assembly = { git = "https://github.com/ai-agent-assembly/python-sdk.git", branch = "master" }

python/agno-tool-policy/src/main.py

@@ -0,0 +1,124 @@
+"""
+agno-tool-policy: Agent Assembly governance demo with Agno (formerly Phidata).
+
+Agno has a **native Agent Assembly adapter**. The SDK governs Agno by patching
+``agno.tools.function.FunctionCall.execute`` — the single chokepoint every Agno
+function-tool call runs through — so every tool an Agno agent invokes passes
+through policy *before* its body executes. A ``deny`` short-circuits the tool
+entirely (its body never runs); an ``allow`` runs it and records the result.
+
+This demo wires the real Agno adapter (``AgnoPatch``) to a local offline policy
+and drives genuine Agno ``@tool`` functions exactly as an Agno ``Agent`` does —
+``FunctionCall(...).execute()`` — so you can watch governance allow the safe
+tools and block the dangerous one with no gateway, API key, or live LLM.
+
+Run (offline mode, no gateway or API key required):
+    uv run python src/main.py
+
+For production use, start the Agent Assembly gateway and update the gateway URL:
+    AGENT_ASSEMBLY_GATEWAY_URL=http://localhost:8080 uv run python src/main.py
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
+
+from agno.tools.function import Function, FunctionCall
+
+from agent_assembly import init_assembly
+from agent_assembly.adapters.agno import AgnoPatch
+
+from src.policy import LocalPolicyEngine
+from src.tools import execute_sql, get_weather, summarize_docs
+
+_BLOCKED_MARKER = "[BLOCKED by governance policy]"
+
+_DEMO_CALLS = [
+    (get_weather, {"city": "London"}),
+    (summarize_docs, {"topic": "policy enforcement"}),
+    (execute_sql, {"sql": "DROP TABLE users; --"}),
+]
+
+
+def _run_governed_call(function: Function, arguments: dict[str, str]) -> None:
+    """Run a real Agno tool through the patched FunctionCall.execute.
+
+    This is the exact call an Agno ``Agent`` makes to execute a tool — the
+    governance hook runs first, so a denied tool returns a failure result and
+    its body never executes.
+    """
+    name = getattr(function, "name", "tool")
+    print(f"  → {name}({arguments})")
+    result = FunctionCall(function=function, arguments=arguments).execute()
+    if result.status == "failure" and _BLOCKED_MARKER in str(result.error):
+        print(f"     ❌ BLOCKED  — {result.error}")
+    else:
+        print(f"     ✅ ALLOWED  — {result.result}")
+    print()
+
+
+def main() -> None:
+    print("=" * 62)
+    print("  Agent Assembly — Agno Tool Policy Demo")
+    print("=" * 62)
+    print()
+
+    gateway_url = os.environ.get("AGENT_ASSEMBLY_GATEWAY_URL", "http://localhost:8080")
+    api_key = os.environ.get("AGENT_ASSEMBLY_API_KEY")
+
+    print(f"Initializing Agent Assembly (gateway: {gateway_url}, sdk-only mode)...")
+
+    with init_assembly(
+        gateway_url=gateway_url,
+        api_key=api_key,
+        agent_id="agno-demo-agent",
+        mode="sdk-only",
+    ) as ctx:
+        print(f"  Agent:    {ctx.client.agent_id}")
+        print(f"  Gateway:  {ctx.client.gateway_url}")
+        print(f"  Mode:     {ctx.network_mode} (offline demo)")
+        print()
+
+        policy = LocalPolicyEngine()
+
+        print("Policy rules (local simulation of gateway policy):")
+        print("  DENY   — execute_sql, run_shell_command  (arbitrary execution)")
+        print("  ALLOW  — everything else")
+        print()
+
+        # In production init_assembly() auto-detects Agno and wires the live
+        # runtime as the interceptor automatically. In this offline sdk-only demo
+        # there is no live runtime, so init_assembly() installs a no-op hook; we
+        # revert it and re-apply the hook wired to our local policy so the demo
+        # shows real allow/deny decisions without a gateway. (The patch is
+        # idempotent, so we must revert the no-op hook before installing ours.)
+        AgnoPatch(policy).revert()
+        patch = AgnoPatch(policy)
+        assert patch.apply(), (
+            "Agno governance hook did not install — is agno importable?"
+        )
+
+        print("Agno governance hook installed on FunctionCall.execute.")
+        print("Tools governed: get_weather, summarize_docs, execute_sql")
+        print()
+
+        print("Running governed tool calls:")
+        print("-" * 44)
+        try:
+            for function, arguments in _DEMO_CALLS:
+                _run_governed_call(function, arguments)
+        finally:
+            patch.revert()
+
+    print("Assembly context shut down.")
+    print()
+    print("Note: in production, init_assembly() auto-detects Agno and applies")
+    print("  this governance automatically — no manual AgnoPatch needed; the")
+    print("  live runtime answers the allow/deny decision instead of the demo policy.")
+
+
+if __name__ == "__main__":
+    main()

python/agno-tool-policy/src/policy.py

@@ -0,0 +1,53 @@
+"""Local offline policy engine for the agno-tool-policy example.
+
+Unlike a framework with no adapter, **Agno has a native Agent Assembly adapter**
+(``agent_assembly.adapters.agno``). The SDK governs Agno by monkey-patching
+``agno.tools.function.FunctionCall.execute`` — the single chokepoint every Agno
+function-tool call runs through — so a governed tool consults this policy's
+``check_tool_start`` *before* its body executes. A ``deny`` short-circuits the
+tool entirely; the body never runs.
+
+This ``LocalPolicyEngine`` simulates the gateway's allow/deny verdicts offline,
+so the demo runs with no gateway or API key. In production the same
+``check_tool_start`` contract is answered by the live runtime instead.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+#: Tools denied by this demo policy — destructive or arbitrary-execution actions.
+DENIED_TOOLS: frozenset[str] = frozenset(
+    {
+        "execute_sql",
+        "run_shell_command",
+    }
+)
+
+
+class LocalPolicyEngine:
+    """Simulates Agent Assembly gateway policy enforcement in offline mode.
+
+    Implements the interceptor contract the Agno adapter calls before running a
+    tool: ``check_tool_start`` returns ``{"status": "allow"}`` or
+    ``{"status": "deny", "reason": ...}``. ``_enforce = True`` puts the engine in
+    the fail-closed posture so an unknown verdict denies rather than allows.
+    """
+
+    _enforce = True
+
+    def check_tool_start(
+        self,
+        *,
+        tool_name: str = "",
+        **kwargs: Any,
+    ) -> dict[str, str]:
+        del kwargs
+        if tool_name in DENIED_TOOLS:
+            return {
+                "status": "deny",
+                "reason": (
+                    f"Tool '{tool_name}' is blocked by policy rule 'deny_arbitrary_execution'."
+                ),
+            }
+        return {"status": "allow"}