Appearance

intutic_clawde.gate.adapters.managed_agents ​

Anthropic Managed Agents (public beta since 2026-04-08) — a session confirmation responder, plus guidance for the two execution surfaces this adapter does NOT cover the same way.

Managed Agents is architecturally different from every other adapter in this package: the customer's process does not run the tool-call loop. Anthropic hosts a "session" that emits an event stream, and the customer's backend answers it. There is no local function call to wrap the way @guard wraps a CrewAI tool.

Verified against a real install, not assumed. anthropic==0.122.0 (the latest published release at the time this adapter was built; the TypeScript twin, @intutic/gate/managed-agents, was checked against @anthropic-ai/sdk@0.117.1 — its latest) was downloaded and its shipped source read directly. Re-read against anthropic==1.11.0 (after the 1.x bump; the 1.x line keeps every beta.sessions name this module and its tests use) — see "1.x additions" below:

  • anthropic/resources/beta/sessions/events.py — every request this package issues (.list(), .send(), .stream()) sends anthropic-beta: managed-agents-2026-04-01 automatically; this module never needs to set that header itself, but {@link MANAGED_AGENTS_BETA} documents the value for callers building their own request.
  • anthropic/types/beta/sessions/beta_managed_agents_user_tool_confirmation_event_params.py — the wire shape this module builds: {"type": "user.tool_confirmation", "tool_use_id", "result": "allow"|"deny", "deny_message"?}. There is no session_thread_id param (0.122.0 and 1.11.0 alike); on the tool-use event it marks a pause cross-posted from a subagent thread and is "Informational only: the server routes the matching user.tool_confirmation ... by tool_use_id, so clients do not send it back." Earlier versions of this module echoed it — removed.
  • anthropic/types/beta/sessions/beta_managed_agents_agent_tool_use_event.py and ..._agent_mcp_tool_use_event.py — both carry evaluated_permission: "allow"|"ask"|"deny"|None. A tool call the server evaluated to "ask" (e.g. because its permission_policy is always_ask — checked live 2026-10-14: an agent created with no permission_policy resolves to always_ask for an MCP toolset and to always_allow for the built-in agent toolset) PAUSES the session — Anthropic's own docs: "Denied tools do not run." A "deny" verdict means the SERVER already denied the call (no confirmation is expected or accepted); "allow" or unset means the call never paused at all. This is the real, documented pre-execution veto this module answers — it is not observe-only.
  • 1.x additions. A third permission_policy, auto (BetaManagedAgentsAutoPolicy): the server judges each invocation. The tool-use events gain evaluation (always_allow | always_ask | auto, the last carrying BetaManagedAgentsAgentAutoEvaluatedPermission — an OPEN union with an open reason_code registry) whose type "always equals the event's top-level evaluated_permission". This module keeps reading only the top-level value: an auto judgement of ask is answered exactly like always_ask, auto allow/deny need no answer, and evaluation is never inspected, so unknown variants cannot change a verdict or crash. A subagent pause cross-posted to the primary stream is answered once (_answered dedups it across list and stream).
  • anthropic/lib/tools/_beta_session_runner.py (SessionToolRunner) is Anthropic's OWN reference dispatcher: it executes agent.tool_use / agent.custom_tool_use calls and posts their results, but it does not decide allow/deny for an ask-gated call itself — it blocks until something sends the matching user.tool_confirmation event. That "something" is what this module provides: {@link confirmation_for_event} / {@link IntuticSessionConfirmer} turn a paused agent.tool_use / agent.mcp_tool_use event into a Gate.guard()-driven verdict.
  • anthropic/lib/environments/_worker.py (EnvironmentWorker, the self-hosted sandbox driver) composes SessionToolRunner internally for built-in agent-toolset tools (bash, edit, read, write, glob, grep, web_fetch, web_search — anthropic/types/beta/beta_managed_agents_agent_tool_config.py). This means self-hosted built-in tools pause and resolve through the IDENTICAL agent.tool_use / evaluated_permission / user.tool_confirmation mechanism as hosted ones — this adapter covers them the same way, contrary to an assumption that a self-hosted sandbox has no governance hook. What differs for self-hosted is only WHERE the tool body executes (the customer's own worker process), not how the pre-execution veto works.
  • agent.mcp_tool_use is explicitly excluded from SessionToolRunner's own dispatch (its own module doc: "the runner never sees a result to post for them" — MCP tools run server-side at Anthropic). But the confirmation gate is a SEPARATE mechanism from execution: BetaManagedAgentsAgentMCPToolUseEvent carries evaluated_permission exactly like the built-in event does, so this module answers MCP pauses too, even though nothing in the official SDK dispatches their execution locally.

## What this module does NOT cover the same way — custom tools

agent.custom_tool_use (beta_managed_agents_agent_custom_tool_use_event.py) has no evaluated_permission field and no permission_policy concept at all: a custom tool always executes wherever the client that owns its name is listening, with no pause. It is answered with user.custom_tool_result (custom_tool_use_id + content + is_error), not user.tool_confirmation. This is the "customer's own process runs the call" surface the phase brief anticipated — the natural integration point IS this package's existing @guard decorator, exactly as the brief guessed. BUT verify the wrapping layer, because a naive port of guard_tools()'s existing .func-swap trick silently does NOT work here:

anthropic.lib.tools._beta_functions.BaseFunctionTool.__init__ captures self._func_with_validate = pydantic.validate_call(func) at CONSTRUCTION time, and BetaFunctionTool.call() invokes self._func_with_validate, not self.func. guard_tools() (framework.py) only ever patches .func in place (the LangChain/smolagents convention) — for a @beta_tool-decorated custom tool that patch is a silent no-op: the gate never runs. The correct pattern is to gate the underlying function BEFORE @beta_tool wraps it, so the validated copy IS the guarded one:

python
from anthropic.lib.tools import beta_tool
from intutic_clawde.gate import guard

@beta_tool
@guard                      # innermost: gate wraps the raw function first
def read_internal_doc(doc_id: str) -> str:
    ...

Decorator order matters: @guard must be the one closest to def.

## What this module does NOT cover at all — the sandbox tool BODY

The self-hosted EnvironmentWorker gates tool PAUSES exactly like the hosted path (see above) — but the tool's actual execution (the real bash/edit/... implementation, anthropic.lib.tools.agent_toolset) runs inside the customer's own worker process, outside this adapter's reach, same as every other Intutic adapter's posture toward a framework's built-in tool bodies (see openai.ts's module doc for the precedent: "no client-side hook exists" is stated plainly rather than pretended otherwise). This module governs WHETHER the call runs; it does not — and cannot — inspect or modify what the tool body does once allowed.

## Webhook vs. streamed/polled delivery — both are real

stream() and list() were exercised against a live session on 2026-10-14; a signed webhook delivery has not been (it needs a public endpoint).

anthropic/resources/beta/webhooks.py: client.beta.webhooks.unwrap(payload, headers=..., key=...) verifies the HMAC signature and returns a typed event; for a pause it is session.requires_action (beta_webhook_session_requires_action_event_data.py — carries only id (the session id), organization_id, workspace_id: NOT the tool call itself). A webhook is therefore a NOTIFICATION to re-poll, not a self-contained payload — {@link IntuticSessionConfirmer.handle_webhook} does exactly that: re-lists the session's unanswered events and answers them. events.stream()/events.list() (resources/beta/sessions/events.py) are the poll/stream alternative this module also supports directly via {@link IntuticSessionConfirmer.poll} / {@link IntuticSessionConfirmer.watch}.

## Fail-closed posture

Matches every other adapter in this package: no gate configured, or an unexpected exception out of Gate.guard(), both produce a "deny" verdict — never a silently-allowed, unevaluated call. Only IntuticGateRefusal is distinguished (its own message becomes deny_message); anything else is wrapped in a generic "gate crashed" deny_message.

Classes ​

IntuticSessionConfirmer ​

python
class IntuticSessionConfirmer(
    client: Any,
    session_id: str,
    *,
    gate: Optional[Gate] = None,
)

Attaches Intutic governance to one Managed Agents session's confirmation pauses.

Duck-typed on client: only client.beta.sessions.events.list(session_id, ...), .send(session_id, events=[...]), and .stream(session_id) are ever called, matching every field/method name in the real anthropic.Anthropic / anthropic.AsyncAnthropic... — this module targets the SYNC client (anthropic.Anthropic), matching this package's synchronous Gate.guard(). No import of anthropic itself: pass any object exposing that shape (a real client, or a test double).

Usage — polling (e.g. a cron/worker loop, or right after creating a session):

python
from anthropic import Anthropic
from intutic_clawde.gate import Gate, GateConfig, install
from intutic_clawde.gate.adapters.managed_agents import IntuticSessionConfirmer

install(Gate(GateConfig()))
client = Anthropic()
confirmer = IntuticSessionConfirmer(client, session.id)
confirmer.poll()   # answers every currently-pending tool_use/mcp_tool_use pause

Usage — webhook (session.requires_action):

python
event = client.beta.webhooks.unwrap(raw_body, headers=request.headers)
if event.data.type == "session.requires_action":
    IntuticSessionConfirmer(client, event.data.id).poll()

Usage — live stream. watch() reconnects with capped backoff and re-polls after every reconnect, so a pause raised while the stream was down is still answered (see watch()'s own doc for the knobs):

python
stop = threading.Event()
for sent in confirmer.watch(stop=stop, idle_timeout_s=120.0):
    print(sent["tool_use_id"], sent["result"])

Methods ​

IntuticSessionConfirmer.handle_event() ​
python
def handle_event(event: Any) -> Optional[Dict[str, Any]]

Answer one event if it needs an answer; returns the sent params, or None if nothing was sent (already answered, or confirmation_for_event returned None — see its doc).

IntuticSessionConfirmer.handle_webhook() ​
python
def handle_webhook(unwrapped_event: Any) -> List[Dict[str, Any]]

Answer the pause a session.requires_action webhook signalled.

unwrapped_event is what client.beta.webhooks.unwrap(raw_body, headers=...) returns (or its .data, or a plain dict of the same shape — see _field's doc). Anything other than a session.requires_action payload is a no-op (returns []): this module only ever answers tool-confirmation pauses, so every other webhook event type is intentionally ignored here.

The webhook payload carries only the session id — it does NOT carry the tool call itself (verified against beta_webhook_session_requires_action_event_data.py) — so this always falls through to a poll().

IntuticSessionConfirmer.poll() ​
python
def poll(*, limit: int = 1000) -> List[Dict[str, Any]]

List recent events and answer every unanswered agent.tool_use/agent.mcp_tool_use pause found. Returns the list of confirmation params sent, in event order.

Filters the listing to the three event types this module reads (types=[...]), so a busy session's message/thinking/usage events never cross the wire for nothing.

Two passes: list() returns events oldest first, so a pause precedes its own user.tool_confirmation. Every confirmation is noted before any pause is answered, or a fresh confirmer (one per webhook delivery) re-answers every pause the session ever had — the live API accepts the duplicate and it wakes an idle session.

IntuticSessionConfirmer.watch() ​
python
def watch(
    *,
    max_reconnects: Optional[int] = None,
    backoff_start_s: float = 0.5,
    backoff_cap_s: float = 10.0,
    idle_timeout_s: Optional[float] = None,
    stop: Optional[Any] = None,
) -> Iterator[Dict[str, Any]]

Catch up on anything already pending, then follow the live event stream, yielding each confirmation as it is sent, reconnecting when the stream drops.

Modelled on anthropic.lib.tools.SessionToolRunner's stream loop: open the stream first, then reconcile with poll(); on end or error sleep (backoff doubling from backoff_start_s to backoff_cap_s, reset on any received event) and reopen. A fatal 4xx (400–499 other than 408 and 429) is re-raised. idle_timeout_s is passed to the SDK as the read timeout, so a silent stream raises APITimeoutError and is reopened. stop is anything with an is_set() (a threading.Event); when set, watch() returns. Ends on a terminal event, stop, or max_reconnects exhausted (unbounded by default).

Functions ​

confirmation_for_event() ​

python
def confirmation_for_event(
    event: Any,
    *,
    gate: Optional[Gate] = None,
) -> Optional[Dict[str, Any]]

Build the user.tool_confirmation params answering event, or None if event needs no answer from this module.

None covers three cases, all correct to send nothing for: the event isn't agent.tool_use/agent.mcp_tool_use (including agent.custom_tool_use — see the module doc); its evaluated_permission is "allow" or unset (the call never paused); or it is already "deny" (the SERVER already resolved it — sending a confirmation for an id the server considers closed would just error).

Anything else — "ask", or a permission value this module doesn't recognise — is treated as a pause needing a verdict, matching SessionToolRunner's own fail-closed handling of unrecognised wire values (hold/deny rather than dispatch).

gate overrides the process-wide instance from intutic_clawde.gate.install(). Pure function: makes no network call itself, so it is the unit-testable core both IntuticSessionConfirmer and a caller with their own event loop can use directly.

Constants ​

MANAGED_AGENTS_BETA ​

python
MANAGED_AGENTS_BETA = 'managed-agents-2026-04-01'

The beta header every Sessions-API request needs. The real SDK (anthropic/resources/beta/sessions/events.py) sets this on every .list()/.send()/.stream() call automatically — this constant exists for callers building their own request against the endpoint directly.

The circuit breaker for AI agents

© 2026 Intutic Community. All rights reserved.