Governing agent frameworks

An IDE has a human watching every step. An unattended LangGraph, LangChain, or CrewAI agent does not. Shrike composes into that agent as four layers — an involuntary floor the model can’t skip, plus the agency it participates through. The result is governance the agent listens to, not a filter bolted to the outside.

LangGraphLangChainCrewAIpip: shrike-guard · npm: shrike-mcp

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.

System promptAgency

Teaches the model what a Shrike verdict means and how to react. One block, set once per session.

Defends against: The model not knowing how to respond to a verdict.

MCP toolsAgency

Gives the model scan channels it can call mid-thought — the model proactively consults Shrike when it is unsure, before it acts.

Defends against: The model choosing not to consult when it should have.

SDK guardFLOOR

Your code scans at the tool boundary before execution. Deterministic — the model cannot skip what your code invokes. The enforcement floor.

Defends against: The model bypassing consultation because "it looked fine."

Block-feedbackAgency

On a block, your code hands the model the rendered reason + recovery on its next turn. The model reads it and adjusts course.

Defends against: The model re-attempting the same blocked action because it does not know what changed.

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.

prompt.py

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.

consult.py

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.

govern.py

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:

feedback.py

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):

agent.py

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
    )

How one risky turn flows

The system prompt (1) has told the model: consult Shrike before a risky action.
Mid-reasoning, the model calls scan_sql_query (2) on its own draft — reads the verdict, rewrites the query.
It runs the rewritten query; the govern() wrapper (3) scans it at the boundary regardless.
If that’s still blocked, format_block_feedback (4) returns the reason + recovery; the model recognizes the shape (from 1) and tries a corrected approach.

CrewAI and other frameworks

The layers are framework-independent. The floor moves to wherever a tool runs — in CrewAI, scan inside _run() before delegating to the real work:

crewai_guard.py

import os
from shrike_guard import ScanClient, format_block_feedback
 
shrike = ScanClient(api_key=os.environ["SHRIKE_API_KEY"])
 
def guarded_run(tool_name, inner_run):
    """Return a _run that scans before delegating to the real tool."""
    def _run(**kwargs):
        verdict = shrike.scan(
            f"tool_call: {tool_name} args={kwargs}",
            context="crewai.tool_boundary",
        )
        action = verdict.get("refuse_tier") or verdict.get("action")
        if not verdict["safe"] and action in ("block", "require_approval"):
            return format_block_feedback(verdict)
        return inner_run(**kwargs)
    return _run

Layer 1 (the system prompt) and Layer 2 (the MCP scan tools) compose the same way — add the block to the agent’s backstory and expose the scan tools to the crew. Any framework that gives you a system prompt and a list of tools supports all four layers.

Why this is guidance, not a filter

—The model participates. With Layers 1 and 2 it consults Shrike on its own, before acting — the agent reasons with the policy, not just against a wall.
—The floor doesn’t depend on cooperation. Layer 3 scans every tool call whether or not the model chose to consult — the guarantee holds on the model’s worst day.
—A block is recoverable. Layer 4 hands back a reason and a path forward in a shape the model was taught to read, so it corrects instead of looping or crashing.
—Context rides across turns. Every verdict carries a session_state block, so the agent’s trajectory is read as a whole — not each call in isolation.