[AAASM-3537] ✨ (adapters): Add Agno framework adapter

README.md

@@ -15,7 +15,7 @@ Python SDK for **AI Agent Assembly** — a governance-native runtime for AI agen
 
 ## Why use it
 
-- **Framework adapters** for LangChain, LangGraph, CrewAI, OpenAI Agents, Pydantic AI, Google ADK, Haystack, Smolagents, and MCP servers — drop in, no SDK rewrites required.
+- **Framework adapters** for LangChain, LangGraph, CrewAI, OpenAI Agents, Pydantic AI, Google ADK, Haystack, Smolagents, Agno, and MCP servers — drop in, no SDK rewrites required.
 - **Pre-execution policy enforcement** via the `FrameworkAdapter` ABC — block disallowed tool calls before they hit the LLM.
 - **Audit trail** — every tool call, prompt, and policy decision is emitted to the gateway with full agent lineage (parent / root / team).
 - **Native PyO3 fast path** (optional) — drop into a Rust runtime client when you need sub-millisecond policy checks.
@@ -41,6 +41,7 @@ are expected to work but are not continuously tested.
 | MCP | `mcp` | `>=1.0.0` | `1.27.x` |
 | OpenAI Agents | `agents` | `>=0.1.0` | `0.17.x` |
 | Smolagents | `smolagents` | `>=1.0.0,<2.0.0` | `1.26.x` |
+| Agno | `agno` | `>=2.0.0` | `2.6.x` |
 
 Python framework compatibility is documented **authoritatively in the SDK docs** —
 [Framework compatibility](./docs/compatibility/frameworks.md) — because the Python

agent_assembly/adapters/agno/__init__.py

@@ -0,0 +1,6 @@
+"""Agno adapter package."""
+
+from agent_assembly.adapters.agno.adapter import AgnoAdapter
+from agent_assembly.adapters.agno.patch import AgnoPatch
+
+__all__ = ["AgnoAdapter", "AgnoPatch"]

agent_assembly/adapters/agno/adapter.py

@@ -0,0 +1,33 @@
+"""Agno (formerly Phidata) framework adapter.
+
+Governs Agno agents at the SDK layer by patching the framework's single
+function-tool execution chokepoint (``FunctionCall.execute`` / ``aexecute``).
+See ``agent_assembly.adapters.agno.patch`` for the hook-point rationale.
+"""
+
+from __future__ import annotations
+
+from agent_assembly.adapters.agno.patch import AgnoPatch
+from agent_assembly.adapters.base import FrameworkAdapter, GovernanceInterceptor
+
+
+class AgnoAdapter(FrameworkAdapter):
+    """Adapter for Agno (pip ``agno``) framework governance hook installation."""
+
+    def __init__(self) -> None:
+        self._patch: AgnoPatch | None = None
+
+    def get_framework_name(self) -> str:
+        return "agno"
+
+    def get_supported_versions(self) -> list[str]:
+        return [">=2.0.0"]
+
+    def register_hooks(self, interceptor: GovernanceInterceptor) -> None:
+        self._patch = AgnoPatch(interceptor)
+        self._patch.apply()
+
+    def unregister_hooks(self) -> None:
+        if self._patch is not None:
+            self._patch.revert()
+            self._patch = None

agent_assembly/adapters/agno/patch.py

@@ -0,0 +1,249 @@
+"""Agno (formerly Phidata) patch module.
+
+Agno (pip ``agno``) runs **every** function-tool body through a single
+chokepoint: ``agno.tools.function.FunctionCall.execute`` (sync) and
+``FunctionCall.aexecute`` (async). An ``Agent`` builds a ``FunctionCall`` from the
+model's tool-call request and invokes ``execute()``/``aexecute()``, which calls
+the user's tool entrypoint and wraps the return in a ``FunctionExecutionResult``.
+
+Patching that one method is therefore the SDK-layer interception point for Agno:
+it sits *before* the tool body runs, so a governance ``deny`` short-circuits the
+body entirely (the tool never executes) and an ``allow`` runs it and records the
+result. This is the same pre-execution allow/deny contract the CrewAI / Pydantic
+AI adapters implement on ``BaseTool.run`` / ``Tool._run``, reusing the shared
+decision-normalization and approval helpers from
+``agent_assembly.adapters.crewai.patch`` so the fail-closed-under-enforce posture
+(AAASM-3107) is identical across adapters.
+"""
+
+from __future__ import annotations
+
+import importlib as importlib
+import inspect
+from dataclasses import dataclass
+from functools import wraps
+from typing import Any
+
+from agent_assembly.adapters.crewai.patch import (
+    _format_approval_rejected_message as _format_approval_rejected,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _format_blocked_message as _format_blocked,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _get_pending_tool_approval_timeout_seconds as _resolve_pending_timeout_seconds,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _interceptor_enforces,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _invoke_sync_tool_check as _invoke_tool_check,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _normalize_decision as _normalize_governance_decision,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _record_sync_tool_result as _record_tool_result,
+)
+from agent_assembly.adapters.crewai.patch import (
+    _wait_for_sync_tool_approval as _wait_for_tool_approval,
+)
+
+_TOOLS_PATCHED_FLAG = "_agent_assembly_agno_functioncall_patched"
+_ORIGINAL_EXECUTE = "_agent_assembly_original_agno_functioncall_execute"
+_ORIGINAL_AEXECUTE = "_agent_assembly_original_agno_functioncall_aexecute"
+
+
+@dataclass(slots=True)
+class AgnoPatch:
+    """Applies Agno runtime monkey-patching hooks on ``FunctionCall.execute``."""
+
+    callback_handler: Any
+
+    def apply(self) -> bool:
+        """Patch ``FunctionCall.execute``/``aexecute`` and report availability.
+
+        Returns:
+            ``True`` when the Agno ``FunctionCall`` class was found and the
+            governance hook installed; ``False`` (a no-op) when ``agno`` is not
+            importable or the expected hook point is absent — never raises
+            ``AttributeError`` on a missing framework.
+        """
+        function_call_cls = _load_agno_functioncall_class()
+        if function_call_cls is None:
+            return False
+        _apply_execute_patch(function_call_cls, self.callback_handler)
+        return True
+
+    def revert(self) -> None:
+        """Revert the Agno ``FunctionCall`` patches when present."""
+        function_call_cls = _load_agno_functioncall_class()
+        if function_call_cls is not None:
+            _revert_execute_patch(function_call_cls)
+        return None
+
+
+def _load_agno_functioncall_class() -> type[Any] | None:
+    try:
+        module = importlib.import_module("agno.tools.function")
+    except ImportError:
+        return None
+
+    function_call_cls = getattr(module, "FunctionCall", None)
+    if isinstance(function_call_cls, type):
+        return function_call_cls
+    return None
+
+
+def _load_agno_failure_result_factory() -> Any | None:
+    """Return a callable building a failure ``FunctionExecutionResult``, or None.
+
+    A governance ``deny`` must short-circuit a tool with a result object that
+    Agno's calling code understands. ``FunctionExecutionResult(status="failure",
+    error=...)`` is exactly what ``execute()`` returns on a failed tool, so the
+    denied message flows back to the model as a tool error rather than a tool
+    result — the body never runs.
+    """
+    try:
+        module = importlib.import_module("agno.tools.function")
+    except ImportError:
+        return None
+    return getattr(module, "FunctionExecutionResult", None)
+
+
+def _build_denied_result(message: str) -> Any:
+    """Wrap a governance block message in an Agno failure result.
+
+    Falls back to returning the raw string when the result type cannot be
+    constructed (e.g. a future Agno that renames it), so a deny still blocks the
+    body even if the wrapper shape changed.
+    """
+    factory = _load_agno_failure_result_factory()
+    if factory is None:
+        return message
+    try:
+        return factory(status="failure", error=message)
+    except Exception:
+        return message
+
+
+def _resolve_tool_name(function_call: Any) -> str:
+    function = getattr(function_call, "function", None)
+    name = getattr(function, "name", None)
+    if isinstance(name, str) and name:
+        return name
+    return str(function_call.__class__.__name__)
+
+
+def _resolve_tool_args(function_call: Any) -> dict[str, Any]:
+    arguments = getattr(function_call, "arguments", None)
+    if isinstance(arguments, dict):
+        return dict(arguments)
+    return {}
+
+
+def _evaluate_decision(
+    callback_handler: Any,
+    *,
+    tool_name: str,
+    tool_args: dict[str, Any],
+    enforce: bool,
+) -> tuple[str, str | None, bool]:
+    """Run the pre-execution governance check and resolve a final verdict.
+
+    Returns ``(status, reason, is_pending_flow)`` where ``status`` is one of
+    ``allow`` / ``deny`` (``pending`` is resolved here by waiting for approval).
+    """
+    decision = _invoke_tool_check(
+        callback_handler,
+        tool_name=tool_name,
+        tool_args=tool_args,
+        agent_id=None,
+    )
+    status, reason = _normalize_governance_decision(decision, enforce=enforce)
+    is_pending_flow = False
+    if status == "pending":
+        is_pending_flow = True
+        timeout_seconds = _resolve_pending_timeout_seconds(callback_handler)
+        final_decision = _wait_for_tool_approval(
+            callback_handler,
+            tool_name=tool_name,
+            timeout_seconds=timeout_seconds,
+            tool_args=tool_args,
+            agent_id=None,
+        )
+        status, reason = _normalize_governance_decision(final_decision, enforce=enforce)
+    return status, reason, is_pending_flow
+
+
+def _apply_execute_patch(function_call_cls: type[Any], callback_handler: Any) -> None:
+    if getattr(function_call_cls, _TOOLS_PATCHED_FLAG, False):
+        return
+
+    original_execute = function_call_cls.execute
+    original_aexecute = getattr(function_call_cls, "aexecute", None)
+    enforce = _interceptor_enforces(callback_handler)
+
+    @wraps(original_execute)
+    def patched_execute(self: Any, *args: Any, **kwargs: Any) -> Any:
+        tool_name = _resolve_tool_name(self)
+        tool_args = _resolve_tool_args(self)
+        status, reason, is_pending_flow = _evaluate_decision(
+            callback_handler,
+            tool_name=tool_name,
+            tool_args=tool_args,
+            enforce=enforce,
+        )
+        if status == "deny":
+            message = _format_approval_rejected(reason) if is_pending_flow else _format_blocked(reason)
+            return _build_denied_result(message)
+
+        result = original_execute(self, *args, **kwargs)
+        _record_tool_result(callback_handler, tool_name=tool_name, result=result)
+        return result
+
+    setattr(function_call_cls, _ORIGINAL_EXECUTE, original_execute)
+    function_call_cls.execute = patched_execute
+
+    if inspect.iscoroutinefunction(original_aexecute):
+
+        @wraps(original_aexecute)
+        async def patched_aexecute(self: Any, *args: Any, **kwargs: Any) -> Any:
+            tool_name = _resolve_tool_name(self)
+            tool_args = _resolve_tool_args(self)
+            status, reason, is_pending_flow = _evaluate_decision(
+                callback_handler,
+                tool_name=tool_name,
+                tool_args=tool_args,
+                enforce=enforce,
+            )
+            if status == "deny":
+                message = _format_approval_rejected(reason) if is_pending_flow else _format_blocked(reason)
+                return _build_denied_result(message)
+
+            result = await original_aexecute(self, *args, **kwargs)
+            _record_tool_result(callback_handler, tool_name=tool_name, result=result)
+            return result
+
+        setattr(function_call_cls, _ORIGINAL_AEXECUTE, original_aexecute)
+        function_call_cls.aexecute = patched_aexecute
+
+    setattr(function_call_cls, _TOOLS_PATCHED_FLAG, True)
+
+
+def _revert_execute_patch(function_call_cls: type[Any]) -> None:
+    if not getattr(function_call_cls, _TOOLS_PATCHED_FLAG, False):
+        return None
+
+    original_execute = getattr(function_call_cls, _ORIGINAL_EXECUTE, None)
+    if callable(original_execute):
+        function_call_cls.execute = original_execute
+
+    original_aexecute = getattr(function_call_cls, _ORIGINAL_AEXECUTE, None)
+    if callable(original_aexecute):
+        function_call_cls.aexecute = original_aexecute
+
+    for attr in (_ORIGINAL_EXECUTE, _ORIGINAL_AEXECUTE, _TOOLS_PATCHED_FLAG):
+        if hasattr(function_call_cls, attr):
+            delattr(function_call_cls, attr)
+    return None

agent_assembly/adapters/registry.py

@@ -5,6 +5,7 @@
 from threading import Lock
 from typing import Callable, Literal
 
+from agent_assembly.adapters.agno.adapter import AgnoAdapter
 from agent_assembly.adapters.base import FrameworkAdapter
 from agent_assembly.adapters.crewai.adapter import CrewAIAdapter
 from agent_assembly.adapters.google_adk.adapter import GoogleADKAdapter
@@ -28,6 +29,7 @@
     "google_adk": 5,
     "haystack": 6,
     "smolagents": 6,
+    "agno": 6,
     "mcp": 99,
 }
 
@@ -72,6 +74,7 @@ def __init__(self) -> None:
             GoogleADKAdapter(),
             HaystackAdapter(),
             SmolagentsAdapter(),
+            AgnoAdapter(),
             MCPAdapter(),
         ]
         for adapter in builtin_adapters:

docs/compatibility/frameworks.md

@@ -52,6 +52,7 @@ to work but are not continuously tested.
 | MCP | `mcp` | `agent_assembly.adapters.mcp` | `>=1.0.0` | `1.27.x` |
 | OpenAI Agents | `agents` | `agent_assembly.adapters.openai_agents` | `>=0.1.0` | `0.17.x` |
 | Smolagents | `smolagents` | `agent_assembly.adapters.smolagents` | `>=1.0.0,<2.0.0` | `1.26.x` |
+| Agno | `agno` | `agent_assembly.adapters.agno` | `>=2.0.0` (patches `FunctionCall.execute` / `aexecute`) | `2.6.x` |
 
 !!! note "Adapter present vs. example present"
     Every framework above has an adapter that is implemented and registered —
@@ -78,6 +79,21 @@ work through the `Tool._run` hook but are not exercised in CI.
 > predated the `>=0.3.0` `AbstractToolset.call_tool` support and is no longer accurate;
 > the example and that pin now track the tested `>=0.3.0` line.
 
+### Agno (formerly Phidata)
+
+Agno (pip `agno`; the package was previously published as `phidata`) routes **every**
+function-tool body through a single chokepoint: `agno.tools.function.FunctionCall.execute`
+(sync) and `FunctionCall.aexecute` (async). An `Agent` builds a `FunctionCall` from the
+model's tool-call request and invokes that method, which runs the user's tool entrypoint
+and wraps the return in a `FunctionExecutionResult`.
+
+The Agno adapter patches that one method, so the pre-execution governance check runs
+*before* the tool body. A `deny` short-circuits the body entirely — the tool never runs —
+and returns a `FunctionExecutionResult(status="failure", ...)` carrying the block message,
+which Agno surfaces to the model as a tool error. An `allow` runs the body and records the
+result. The **tested line is `agno>=2.0.0`** (currently `2.6.x`), installed by the SDK's
+dev/test dependency group and exercised by the `importorskip`-guarded integration test.
+
 ## Declaring frameworks in your own project
 
 The frameworks above are **not** runtime dependencies of `agent-assembly` — installing

pyproject.toml

@@ -88,6 +88,13 @@ test = [
     # tests drive a real `Tool` subclass, so a regression to a fail-open no-op
     # patch is caught (deny must block `forward`) instead of silently skipped.
     "smolagents>=1.0.0,<2.0.0",
+    # AAASM-3537: dev/test-only (NOT a runtime dependency). Agno (formerly
+    # Phidata) runs every function-tool body through
+    # `agno.tools.function.FunctionCall.execute` / `aexecute`. Installing it lets
+    # the `importorskip`-guarded integration test drive a real `FunctionCall`, so
+    # the negative-control deny test fails if the patch ever regresses to a no-op
+    # instead of being silently skipped.
+    "agno>=2.0.0",
 ]
 pre-commit-ci = [
     "pre-commit>=3.5.0,<5",