Plugging Compiled into your agents

Compiled inspects what an agent is about to do — a tool call, an action, an outbound message — and returns allow / flag / block before it runs.

Every adapter takes a guard, which is one of two things:

• In-process (AgentGuard) — the detector runs inside your process for the lowest-latency, zero-egress hot path.
• Remote (GuardClient) — calls a Compiled scanner's /guard/action over HTTP, for agents that don't run the detector locally.

Both expose .check(...) and return the same decision, so the adapters below are identical either way.

The guard

# In-process (lowest latency)
from compiled.scanner import FabricScanner
from compiled.agent_guard import AgentGuard
fs = FabricScanner("library/manifest.json")
guard = AgentGuard(fs.fabric, fs.embed)

# …or remote
from compiled.integrations import GuardClient
guard = GuardClient("https://compiled.your-tenant.internal", fail_closed=True)

decision = guard.check(tool="send_email", arguments={"to": "x@y.com", "body": "…"})
if not decision.allowed:
    raise RuntimeError(decision.reason)   # blocked

OpenAI tool calls

Guard the tool calls the model proposes, and only dispatch the allowed ones:

from compiled.integrations.openai_tools import guarded_openai_dispatch

msg = client.chat.completions.create(model="gpt-4.1", messages=msgs, tools=tools)
tool_msgs = guarded_openai_dispatch(
    guard, msg.choices[0].message.tool_calls, dispatch=run_my_tool)
msgs += tool_msgs   # blocked calls come back as a tool message, never executed

Anthropic (Claude) tool use

from compiled.integrations.anthropic_tools import guarded_anthropic_dispatch

resp = client.messages.create(model="claude-opus-4-8", messages=msgs, tools=tools)
results = guarded_anthropic_dispatch(guard, resp.content, dispatch=run_my_tool)
# blocked tool_use blocks return an is_error tool_result the model can react to

LangChain

from compiled.integrations.langchain_tool import guard_langchain_tool
safe_tool = guard_langchain_tool(guard, my_tool)   # drop-in; .invoke is guarded

MCP

Front your MCP server with the gateway so every tools/call is checked before it reaches the tool:

from compiled.mcp_gateway import MCPGuardGateway
gateway = MCPGuardGateway(guard)

Any HTTP agent endpoint (ASGI)

Guard an action endpoint with one line of middleware:

from compiled.integrations.asgi import GuardMiddleware
app.add_middleware(GuardMiddleware, guard=guard, paths={"/agent/act"})
# matched POSTs are parsed, checked, and blocked with 403 before your handler

Any language (REST)

POST /guard/action
{ "tool": "transfer_funds", "arguments": { "amount": 50000, "to": "…" } }
→ { "decision": "block", "allowed": false, "reason": "…", "risk": 0.91 }

Failure mode

GuardClient(fail_closed=True) blocks actions when the guard is unreachable; the default fails open (allow) so a guard outage never halts your agents. Choose per the risk of the action you're guarding — a transfer_funds tool should fail closed; a read-only lookup can fail open.

Answer & action reuse (agent memory)

The same in-tenant embedder that decides allow / flag / block also answers "has an agent already done this?" — the positive polarity of the substrate. Before an agent runs an expensive or risky action, ask the memory:

The memory endpoints are authenticated (set COMPILED_AGENT_TOKEN) — a write seeds a cache other agents trust, so it must not be open. A reuse hit requires an exact canonical-text match (not just embedding similarity), so a near-match forgery can't be served as a trusted result; entries carry provenance and an optional TTL (COMPILED_MEMORY_TTL), and the index is bounded (COMPILED_MEMORY_MAX).

from compiled.integrations import GuardClient
guard = GuardClient("https://compiled.your-tenant.internal", token="<COMPILED_AGENT_TOKEN>")

hit = guard.recall(tool="lookup_refund_policy", arguments={"region": "EU"})
if hit["status"] == "reuse":
    result = hit["result"]          # exact match — skip the re-run entirely
elif hit["status"] == "surface":
    show_user(hit["match"])         # near match — "someone already solved this"
else:
    result = run_the_tool(...)      # miss — do the work, then remember it
    guard.remember(tool="lookup_refund_policy", arguments={"region": "EU"}, result=result)

• reuse (similarity ≥ 0.95): return the cached result, skip the action.
• surface (≥ 0.85): advisory — surface the prior answer to the human/agent.
• miss: do the work, then remember it for next time.

Thresholds are tunable (COMPILED_REUSE_THRESHOLD / COMPILED_SURFACE_THRESHOLD). Endpoints: POST /agent/recall and POST /agent/remember. Everything is embedded and stored in-tenant — the memory never leaves your environment. This dedupes repeated agent work, caches answers, and surfaces cross-agent intelligence ("someone just asked this").

See examples/agent_integrations.py and examples/agent_memory.py for runnable, offline demos.