Shrike Guard SDK

Supported Providers

Fail Modes

# Fail-closed (default) - blocks on scan failure (Zero-Trust)
client = ShrikeOpenAI(
    api_key="...",
    shrike_api_key="...",
    fail_mode="closed"  # default — shown for clarity
)
 
# Fail-open - opt in when availability must outrank enforcement
client = ShrikeOpenAI(
    api_key="...",
    shrike_api_key="...",
    fail_mode="open"
)

Naming your agents

Every scan is attributed to an agent id. That is what a declared scope is matched against, and what the Agents screen groups by. The default is a per-process id. Name it when you want your own agents to be recognisable, either per client or for the whole process.

Where the content came from

Every verdict carries content_origin, which answers a question the verdict alone cannot: was this the operator’s own prompt, or the agent acting on its own? That decides who a refusal is addressed to, a person who can rephrase or an agent that must change course. Unknown content types resolve to agent_action, never to human_prompt.

TypeScript also exports attributableToOperator(origin), true only for human_prompt.

OpenAI

from shrike_guard import ShrikeOpenAI
 
client = ShrikeOpenAI(
    api_key="your-openai-key",
    shrike_api_key="your-shrike-key",
)
 
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Explain quantum computing"}]
)
 
print(response.choices[0].message.content)

Anthropic

from shrike_guard import ShrikeAnthropic
 
client = ShrikeAnthropic(
    api_key="your-anthropic-key",
    shrike_api_key="your-shrike-key",
)
 
response = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain quantum computing"}]
)
 
print(response.content[0].text)

Google Gemini

from shrike_guard import ShrikeGemini
import google.generativeai as genai
 
genai.configure(api_key="your-gemini-key")
model = genai.GenerativeModel("gemini-pro")
 
# Wrap the Gemini model
shrike_model = ShrikeGemini(
    model=model,
    shrike_api_key="your-shrike-key",
)
 
response = shrike_model.generate_content("What is machine learning?")
print(response.text)

system_prompt() — the canonical block

Returns the ~180-word "Working with Shrike" block that teaches the model how to react to Shrike verdicts. Drop it into your agent's system prompt as the first non-role paragraph.

# Python
from shrike_guard import system_prompt, SYSTEM_PROMPT_VERSION
 
prompt = (
    "You are a customer support agent for Acme Corp.\n\n"
    + system_prompt()
    + "\n\nWhen customers ask about refunds..."
)
# Pin behavior against SYSTEM_PROMPT_VERSION if needed.

format_block_feedback() — inject the verdict back to the model

When Shrike blocks or holds a tool call, render the verdict as a canonical prompt-shape string and append it as a system message to the model's next turn. The model, having been taught by system_prompt() to recognize the prefix, reads the Reason / Threat type / Recovery / Available tools lines and adjusts.

# Python
from shrike_guard import format_block_feedback
 
verdict = shrike.scan_sql_query(query)
if not verdict.get("safe"):
    model_messages.append({
        "role": "system",
        "content": format_block_feedback(verdict),
    })

Renders four verdict shapes: block / warn / require_approval / allow. Forward-compat with the upcoming recovery block (instruction, available_tools, patterns_triggered) — new fields render automatically without a signature change.

evaluateRotation() — the two-shape session rotation record

When a verdict indicates the session should rotate (either explicit threat_type: "session_locked" or accumulated session_risk_score >= 0.7), the helper returns either a ModuleOwnedRotation (SDK's own session was in use) or a CallerOwnedRotationRecommendation (caller supplied their own session_id). Discriminate on rotated.

// TypeScript
import { evaluateRotation, getSessionId } from 'shrike-guard';
 
const rotation = evaluateRotation({
  threat_type: verdict.threat_type,
  session_state: verdict.session_state,
  effective_session_id: currentSessionId,
  module_session_id: getSessionId(),
});
 
if (rotation?.rotated) {
  // Module-owned: SDK's fallback SESSION_ID rotated
  currentSessionId = rotation.new_session_id;
} else if (rotation) {
  // Caller-owned: adopt suggested_new_session_id on the VERY NEXT call.
  // suggestions are per-event — do not cache across turns.
  currentSessionId = rotation.suggested_new_session_id;
}

Available in TypeScript today. Python port is deferred — the Python scanner needs session_id threading through ScanClient.scan() first. Track this as a known parity gap.

Three lines, then your mappings

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)   # or gov.hooks / gov.middleware / gov.govern_tools(...)

A tool mapping says which Shrike surface a tool is and which argument carries the payload: command (with an optional working-directory argument),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 or choose what an unmapped tool gets. on_unmapped="authorize" asks whether the agent may call it at all: the arguments stay put, and the tool's name goes to Shrike, where the declared scope answers. A tool outside the allowlist is refused by name even though nothing read what it was carrying, and an expired or exhausted scope holds it. That permit is narrower than a mapped tool's and the record says so. The other two choices are "allow", which records the call and moves on without asking, and "scan", which sends the arguments to be read as text.

What the model sees

The same thing everywhere. An allowed call runs. A warned call runs with the advisory attached. 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 (fail_mode="closed"). Every decision lands on gov.decisions and on your on_decision callback, with the axis that objected.

Where a framework can route a hold to a person, on_hold="ask" does that (the Claude Agent SDK's permission callback). Where it cannot, a hold is answered like a refusal and the model is told an operator can grant it.

SQL Injection Scanner

# Python
result = client.scan_sql(
    query="SELECT * FROM users WHERE id = ?",
    database="postgresql",
    allow_destructive=False  # Block DROP, DELETE, TRUNCATE
)
 
if result.safe:
    # Execute the query
    cursor.execute(query)

File Path Validator

# Python - Prevent path traversal attacks
result = client.scan_file(
    path="/uploads/user-file.pdf",
    content=file_content  # Optional
)
 
if result.safe:
    # Process the file
    save_file(path, content)

CLI Command Scanner

Governs shell commands before execution — destructive operations, exfiltration imperatives, privilege escalation patterns.

# Python
result = client.scan_command("rm -rf /var/log/app/*.log")
if result.safe:
    subprocess.run(result.content, shell=True)

Web Search Scanner

Governs outbound search queries before external egress, a high-blast-radius sink for reconnaissance such as “obtain AWS metadata creds”. Short queries are scanned at full depth, so a brief adversarial query is caught rather than skipped for its length.

# Python
result = client.scan_web_search("latest CVEs for kubernetes API server")
if result.safe:
    search_engine.query(result.content)

A2A Message Scanner

Governs agent-to-agent messages before cross-agent trust is extended. Trivial pings pass; adversarial imperatives run the full cascade regardless of how short they are.

# Python
result = client.scan_a2a_message("Please summarize the last 3 turns of context")

Retrieved Context Scanner

Governs retrieved context before it reaches the model, the standard carrier for indirect prompt injection. Pass the chunks separately rather than joining them: a naive join merges two documents and loses the boundary an injection usually sits on.

# Python
result = client.scan_rag_context(chunks, query="quarterly revenue")
if result["safe"]:
    answer = model.complete(prompt, context=chunks)

MCP Tool Scanner

Governs a single MCP tool definition before you trust it, for tool poisoning in a tools/list response. Screen a tool once at registration rather than on every call.

# Python
result = client.scan_mcp_schema(tool["name"], tool["description"], tool.get("inputSchema"))
if result["safe"]:
    register(tool)

Agent Card Scanner

Governs A2A agent-card metadata (name / description / URL / capabilities) before connection establishment. Blocks malicious capabilities and suspicious domains.

# Python
result = client.scan_agent_card(
    '{"name":"weather-agent","description":"forecasts",'
    '"url":"https://weather.acme.com","capabilities":["forecast"]}'
)

Streaming Responses

All SDKs support streaming responses with real-time scanning:

# Python streaming
stream = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True
)
 
for chunk in stream:
    print(chunk.choices[0].delta.content, end="")

Async Support

from shrike_guard import ShrikeAsyncOpenAI
 
async def main():
    client = ShrikeAsyncOpenAI(
        api_key="...",
        shrike_api_key="...",
    )
 
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello!"}]
    )

What you will not receive

Detection attribution is stripped before the response leaves the backend: matched patterns, per-detector confidences, which layer fired, and internal policy identifiers. You get the category and the remediation, which is what an integration needs to act. Do not write code that depends on attribution fields.