Governing agent frameworks

The starters: three lines per framework

Every starter has the same shape. Build it from a ScanClient, map your tools to Shrike’s surfaces, hand it to the framework. The core is the same module underneath; only the last line changes.

from agents import Agent
from shrike_guard import ScanClient
from shrike_guard.openai_agents import govern   # or claude_agent, google_adk, langgraph_agent, crewai_agent
 
guard = ScanClient(api_key=SHRIKE_KEY, agent_id="invoice-agent", session_id=run_id)
gov = govern(guard, agent_id="invoice-agent")
gov.declare(["file_path", "file_content", "sql"], purpose="Reconcile Q3 vendor invoices")
 
gov.map_tool("run_query", "sql", arg="query")
gov.map_tool("save_report", "file", path_arg="path", content_arg="text")
gov.exempt("get_time")
 
agent = gov.govern_agent(Agent(name="invoices", tools=[run_query, save_report, get_time]))

A tool mapping says which surface a tool is and which argument carries the payload:command, file (a path and the content written to it), file_path,sql, web_search, rag_context, a2a_message, agent_card, or none for a tool allowed without a scan. The Claude Agent SDK starter ships mappings for the SDK’s built-in tools; the others govern the tools you write, so they need your mappings. A tool with no mapping is refused, with a message that says how to map it, until you map it.

What the model sees is the same everywhere. An allowed call runs. A warned call runs with the advisory. A blocked call does not run and the model gets the reason. A held call, an action outside the declared scope, does not run and the model gets the reason, the recovery, and the instruction to ask with request_scope and then stop; an operator decides on the Agents screen. An action Shrike could not check is refused. Every decision lands on the record with the axis that objected, content or authorization.

Install: pip install shrike-guard[openai-agents] (or claude-agent,google-adk, langgraph, crewai, frameworks for all);npm install shrike-guard plus the framework package and zod.

Each package name above links to the module that implements it. To watch a governed agent get refused before you wire one up, the playground runs a scripted agent whose every tool call is scanned for real, approval pauses included.

The stack — a floor plus agency

Each layer defends against a different failure mode. Layer 3 alone is enforcement — it guarantees nothing gets past the tool boundary unscanned. Layers 1, 2, and 4 turn that enforcement into guidance: the model reads Shrike’s context, consults on its own, and corrects course. You ship all four.

Layer 3 by itself is a filter. Layers 1 + 2 + 3 + 4 together is the active guidance layer — the model chooses correctly because it has read Shrike’s context, and can’t act wrongly even when it doesn’t.

Layer 1 — the system prompt

Teach the model how to work with Shrike. The SDK ships the canonical “Working with Shrike” block — drop it in as the first non-role paragraph of your agent’s system prompt. This is what makes the later layers legible: the model recognizes a verdict and knows a block is a course-correction, not a dead end.

from shrike_guard import system_prompt
 
ROLE = "You are a data-ops agent that answers questions over the warehouse."
 
# Layer 1: the model now knows what a verdict means and how to react.
AGENT_PROMPT = system_prompt() + "\n\n" + ROLE

Layer 2 — MCP tools the model consults

Give the model Shrike’s scan channels as callable tools. Prompted by Layer 1, the model invokes them mid-thought — “before I run this query, let me check it” — and adjusts based on the verdict. This is the agentic half: the model participating in its own governance, not just being filtered.

from langchain_mcp_adapters.client import MultiServerMCPClient
 
# Expose Shrike's scan channels so the model can self-consult.
shrike_mcp = MultiServerMCPClient({
    "shrike": {"command": "npx", "args": ["-y", "shrike-mcp"], "transport": "stdio"},
})
 
# scan_prompt, scan_sql_query, scan_command, scan_web_search, scan_file_write,
# scan_response, scan_a2a_message, scan_agent_card, + session_status and more.
consult_tools = await shrike_mcp.get_tools()

The MCP server exposes 14 tools in total — 8 scan channels plus utilities like session_status (read the accumulated session risk without acting) and scan_declare_scope (tell Shrike what this agent is allowed to do). See the MCP guide.

Layer 3 — the SDK guard (the floor)

Consultation is voluntary; the floor is not. Wrap each real tool so every call is scanned before it runs — the model invokes the tool normally and cannot see or skip the wrapper. On a block, hand back the rendered block-feedback (Layer 4) instead of executing, so the model gets a structured way to recover rather than a crash.

import os
from shrike_guard import AsyncScanClient, format_block_feedback
from langchain_core.tools import BaseTool, StructuredTool
 
shrike = AsyncScanClient(api_key=os.environ["SHRIKE_API_KEY"])
 
def govern(tool: BaseTool) -> BaseTool:
    """Wrap a tool so every call is scanned before it runs.
    The model invokes the tool normally — it can't see or skip this."""
    async def guarded(**kwargs):
        verdict = await shrike.scan(
            f"tool_call: {tool.name} args={kwargs}",
            context="langgraph.tool_boundary",
        )
        action = verdict.get("refuse_tier") or verdict.get("action")
        if not verdict["safe"] and action in ("block", "require_approval"):
            # Layer 4: render reason + recovery in the shape Layer 1 taught the model to read.
            return format_block_feedback(verdict)
        return await tool.ainvoke(kwargs)
 
    return StructuredTool.from_function(
        coroutine=guarded,
        name=tool.name,
        description=tool.description,
        args_schema=tool.args_schema,
    )

Layer 4 — block-feedback the model reads

When the floor blocks, the model needs to know why and what to do instead, or it will just retry the same action. format_block_feedback(verdict) renders the category and recovery guidance in the exact shape the Layer 1 block taught the model to recognize — so the next turn is a corrected attempt, not a repeat. It’s already wired into the govern() wrapper above; you can also call it directly anywhere you enforce:

from shrike_guard import format_block_feedback
 
verdict = shrike.scan(user_sql, context="tool.sql")
action = verdict.get("refuse_tier") or verdict.get("action")
if not verdict["safe"] and action == "block":
    # Append as a system/tool message on the model's next turn.
    messages.append({"role": "system", "content": format_block_feedback(verdict)})

Putting it together in one agent

The whole integration is one file. Four layers, annotated inline — the system prompt (1), the scan tools the model can call (2), the floor that wraps every real tool (3), and the block-feedback it returns on a block (4):

import os
from shrike_guard import AsyncScanClient, system_prompt, format_block_feedback
from langchain_core.tools import BaseTool, StructuredTool
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
 
shrike = AsyncScanClient(api_key=os.environ["SHRIKE_API_KEY"])
 
def govern(tool: BaseTool) -> BaseTool:                     # Layer 3: floor (+ Layer 4 on block)
    async def guarded(**kw):
        v = await shrike.scan(f"tool_call: {tool.name} args={kw}", context="langgraph.tool")
        if not v["safe"] and (v.get("refuse_tier") or v.get("action")) in ("block", "require_approval"):
            return format_block_feedback(v)                 # Layer 4: reason + recovery back to model
        return await tool.ainvoke(kw)
    return StructuredTool.from_function(coroutine=guarded, name=tool.name,
                                        description=tool.description, args_schema=tool.args_schema)
 
async def build_agent(model):
    npx = lambda *a: {"command": "npx", "args": ["-y", *a], "transport": "stdio"}
    consult = await MultiServerMCPClient({"shrike": npx("shrike-mcp")}).get_tools()          # Layer 2
    real    = await MultiServerMCPClient({"github": npx("@modelcontextprotocol/server-github")}).get_tools()
    return create_react_agent(
        model,
        tools=consult + [govern(t) for t in real],          # agency + floor
        prompt=system_prompt() + "\n\nYou are a data-ops agent.",   # Layer 1
    )

CrewAI and other frameworks

The layers are framework-independent, and the floor is the starter. In CrewAI the starter wraps each tool so Shrike judges the call before the inner tool runs and the model reads the reason as the tool’s result; a global hook covers tools you cannot reach:

import os
from crewai import Agent
from shrike_guard import ScanClient
from shrike_guard.crewai_agent import govern
 
guard = ScanClient(api_key=os.environ["SHRIKE_API_KEY"], agent_id="clerk", session_id=run_id)
gov = govern(guard, agent_id="clerk").map_tool("run_query", "sql", arg="query")
 
clerk = Agent(role="Invoice clerk", goal=GOAL, backstory=BACKSTORY, tools=gov.govern_tools([run_query]))
uninstall = gov.install()   # the global before_tool_call backstop and the observe hook

Layer 1 (the system prompt) and Layer 2 (the MCP scan tools) compose the same way in any framework: add the block to the agent’s instructions and expose the scan tools. A framework without a starter yet still supports all four layers through the SDK guard.

Why this is guidance, not a filter