intutic_clawde.gate.adapters.ag2

AG2 adapter: IntuticMiddleware, plus make_intutic_middleware for an explicit Gate.

Read this before trusting the "AG2 = pre-Microsoft AutoGen fork" framing in the plan that scoped this adapter. That framing is stale. The ag2 package on PyPI was substantially rewritten — ag2==1.0.2 (installed live to write this adapter) no longer imports as autogen at all and shares no API surface with the ConversableAgent/GroupChat shape most AG2/pyautogen tutorials still describe; it is a from-scratch architecture built around typed Events flowing through a Stream, agents (ag2.Agent) driven by a Context, and a BaseMiddleware chain intercepting on_turn/on_llm_call/ on_tool_execution/on_human_input. The veto mechanism the plan predicted — "subclass BaseMiddleware, override on_tool_execution, veto by not calling call_next" — turned out to be exactly right regardless, confirmed by reading ag2/middleware/base.py, ag2/tools/executor.py, and ag2/tools/final/function_tool.py directly, not by trusting the docs:

• BaseMiddleware.on_tool_execution(self, call_next, event, context) is a real method; the base implementation is exactly return await call_next(event, context), i.e. the identity middleware — overriding it and returning a synthetic result INSTEAD of calling call_next is how FunctionTool.register documents and constructs its own chain (_wrap_middleware(mw.on_tool_execution, execution), folded right to left over self._middleware/the middleware=[...] passed to Agent(...)).
• event: ToolCallEvent carries event.name (the tool name) and event.serialized_arguments — a lazily-JSON-decoded dict property over the raw event.arguments string, so no manual json.loads is needed here (unlike the OpenAI Agents SDK / AutoGen adapters, whose SDKs hand over the raw string).
• the return type is ToolResultEvent | ToolErrorEvent | ClientToolCallEvent (ag2.middleware.base.ToolResultType), not a bare ToolResult — a veto uses ToolErrorEvent.from_call(event, exc), AG2's own constructor for "this tool call failed," which the tool executor's error path already knows how to surface to the model. Confirmed live: constructing one from a real ToolCallEvent and a plain exception round-trips through ToolResult/TextInput correctly with no other setup required.
• Agent(..., middleware=[IntuticMiddleware]) — the class itself, not an instance — is the documented shape: MiddlewareFactory is Protocol.__call__(self, event, context) -> BaseMiddleware, which a BaseMiddleware subclass's own constructor satisfies. Passing the class resolves the gate via intutic_clawde.gate.active() at call time (the process-wide gate); make_intutic_middleware(gate=...) below returns a small factory closing over a specific Gate instance for tests or multi-gate processes, the same shape openai_agents.py's make_intutic_tool_guardrail uses.

Now confirmed live: what AG2's own event/stream dispatcher does with an exception escaping on_tool_execution itself (as opposed to the ToolErrorEvent this adapter returns deliberately) — driven through AG2's REAL dispatcher (ag2.tools.executor.ToolExecutor wired to a real MemoryStream/ConversationContext, the same construction ag2.plugin's Agent setup uses internally), not a hand-rolled call. See tests/test_adapter_ag2.py's TestRealDispatcher class, in particular test_an_exception_escaping_middleware_is_caught_by_ag2s_own_dispatcher_not_swallowed_as_allowed, which uses a bare middleware with NO catch-all at all. Unlike CrewAI (confirmed fail-OPEN — swallows any non-HookAborted exception and reports the call as allowed; see crewai.py's module doc), AG2 fails CLOSED: ag2/tools/executor.py's _execute_call wraps await context.send(call) in a bare except Exception that converts ANY exception escaping a middleware's on_tool_execution into a real ToolErrorEvent.from_call(call, e), with the underlying tool body never invoked. IntuticMiddleware's own defensive catch-all (converting every exception, not only IntuticGateRefusal, into a ToolErrorEvent itself) is therefore CONFIRMED NOT strictly necessary for AG2's dispatcher specifically — AG2 would already fail closed even without it. It is kept anyway, deliberately: without it, a "no gate configured" misconfiguration or a Gate.guard() bug would still surface as a ToolErrorEvent, but with AG2's own auto-generated traceback text rather than a [Intutic Governance] BLOCKED: ...-prefixed, model-legible message an operator/dashboard can recognize — the same reasoning crewai.py's catch-all carries, just for a different (cosmetic, not correctness) reason on this framework.

Optional import: importing this module never fails even without ag2 installed. Only instantiating IntuticMiddleware (or calling make_intutic_middleware) requires it — pip install intutic-clawde[ag2].

Classes

IntuticMiddleware

class IntuticMiddleware(event: Any, context: Any, *, gate: Optional[Gate] = None)

Bases: BaseMiddleware

AG2 BaseMiddleware — vetoes a tool call before it runs.

Usage:

from ag2 import Agent
from intutic_clawde.gate.adapters.ag2 import IntuticMiddleware

agent = Agent("my-agent", middleware=[IntuticMiddleware])

Pass the CLASS (a MiddlewareFactory), not an instance — AG2 constructs one per turn via IntuticMiddleware(event, context). Resolves the process-wide gate (intutic_clawde.gate.active()) at call time. For an explicit Gate (tests, multi-gate processes), use make_intutic_middleware(gate=...) instead.

Attributes

Name	Type	Default	Description
on_tool_execution	—	_make_on_tool_execution(None)

Functions

make_intutic_middleware()

def make_intutic_middleware(*, gate: Optional[Gate] = None)

Builds a MiddlewareFactory bound to a specific Gate.

Usage:

agent = Agent("my-agent", middleware=[make_intutic_middleware(gate=my_gate)])

Prefer the module-level IntuticMiddleware class unless a test or a multi-gate process needs to bind a specific Gate instance rather than the process-wide one installed via intutic_clawde.gate.install().