shield-kya

In-process kya (Know Your Agent) governance shim for Python agent frameworks.

shield_kya delegates policy evaluation to the local kya CLI (kya wrap --offline --json ...), so Python tools share the same offline verdict path and trail as the TypeScript SDK. Stdlib only, no dependencies.

Requirements

• Python >= 3.9 (the framework hooks need what their framework needs: langchain 3.10+, pydantic-ai 3.11+, crewai 3.10 to 3.13, google-adk 3.10+, agent-framework-core 3.10+, openai-agents 3.10+)
• The kya CLI on PATH (npm i -g @shield-agent/kya), or set KYA_CLI to an absolute path (an executable, or a .js file run with node). kya is looked up in absolute PATH entries only, never the current directory; a relative KYA_CLI is ignored.

Install

Install from PyPI:

pip install shield-kya

To install this checkout instead, run this in this directory:

pip install .
pip install '.[langchain]'     # LangChain / LangGraph middleware
pip install '.[pydantic-ai]'   # Pydantic AI capability
pip install '.[crewai]'        # CrewAI global hook
pip install '.[adk]'           # Google ADK plugin
pip install '.[agent-framework]'  # Microsoft Agent Framework middleware (agent-framework-core)
pip install '.[openai-agents]'   # OpenAI Agents SDK tool input guardrail

Semantics: advisory, fail-open

Same rules as kya hook:

Without a CLI verdict the never-list is matched in Python (shield_kya/_never_list.py, a port of the CLI's name-only rule; tests/test_never_list.py and the TS suite assert the same vectors file). KYA_FAIL_CLOSED accepts 1/true/yes/on (any case). Hosted operators should not pass fail_closed=False, which overrides the env var.

On Windows the shim refuses to run kya through a .cmd/.bat shim (cmd.exe would re-parse the model-controlled arguments); set KYA_CLI to the absolute path of the package's dist/cli.js and it runs with node.

Arguments are passed to the CLI on argv. A tool call whose arguments exceed the OS argv limit (about 128 KiB per argument on Linux) gets no CLI verdict and no trail row: it fails open, and the never-list still applies by tool name. An empty tool id is denied (EMPTY_TOOL), like the CLI's offline evaluate.

"Hosted" here means only the fail-closed switch: the shim always evaluates offline (kya wrap --offline) and never consults a hosted policy plane.

Async callers use an asyncio subprocess, so the check never blocks the event loop.

LangChain / LangGraph (agent-wide)

from langchain.agents import create_agent
from shield_kya.integrations.langchain import KyaMiddleware

agent = create_agent(model, tools, middleware=[KyaMiddleware()])

Each tool call this agent dispatches is checked before it runs (one trail row per call that the CLI evaluated). A DENY skips the tool and returns an error ToolMessage with the reason (KYA denied postgres__drop_table: NEVER_EVENT), so the model sees it and the run continues. It covers tool calls dispatched by that agent's ToolNode. Not covered: subagents or graphs built separately (add the middleware there too), direct tool.invoke() calls, tools the model provider executes server-side, and MCP tools whose names carry no server prefix, so the never-list cannot recognize them (server= applies one prefix to every tool of the agent). Put KyaMiddleware last in middleware (innermost) so it checks the call as it will execute, after any middleware that rewrites tool calls. Options: server= (tool id becomes server__tool), irreversible=, strict= (raise KyaDenied, and block REQUIRE_APPROVE), fail_closed=, timeout=.

Raw LangGraph graph:

kya = KyaMiddleware()
ToolNode(tools, wrap_tool_call=kya.wrap_tool_call, awrap_tool_call=kya.awrap_tool_call)

Pydantic AI (agent-wide)

from pydantic_ai import Agent
from shield_kya.integrations.pydantic_ai import KyaCapability

agent = Agent(model, tools=[...], capabilities=[KyaCapability()])

The capability hooks before_tool_execute, so every function tool the agent runs (toolsets and MCP servers included, but see Known limits on MCP names) is checked before it executes, with the CLI run as an asyncio subprocess (run_sync included, since Pydantic AI runs tools on its event loop). A DENY raises SkipToolExecution with the reason, so the model receives KYA denied postgres__drop_table: NEVER_EVENT as the tool result and the run continues. strict=True raises KyaDenied instead, which aborts the run, and also blocks REQUIRE_APPROVE. Output tools skip capability hooks and are not checked. The capability is per agent: add KyaCapability() to every agent, including agents another agent runs as a tool (subagents). Put KyaCapability last in capabilities so the trail records the args as they will execute. Options: server=, irreversible=, strict=, fail_closed=, timeout=.

CrewAI (process-wide)

from shield_kya.integrations import crewai as kya_crewai

kya_crewai.register()   # every tool call of every crew in this process
crew.kickoff()
kya_crewai.unregister() # optional

register() adds a global PRE_TOOL_CALL hook and a matching POST_TOOL_CALL hook; calling it again replaces them. CrewAI swallows every hook exception except HookAborted and then runs the tool, so a deny is always a HookAborted: the tool does not run, and the post hook replaces CrewAI's fixed "Tool execution blocked by hook" text with the KYA reason, so the agent sees it and the crew continues. For the same reason strict=True blocks REQUIRE_APPROVE but never raises KyaDenied, and an unexpected error inside the hook still goes through the fail-open / never-list / fail-closed rules instead of reaching CrewAI. The tool id is the tool's own name: CrewAI's LLM-facing name is sanitized (postgres__drop_table becomes postgres_drop_table), which the never-list would not recognize. Call register() after your own pre-tool hooks so the check sees their input edits. Options: server=, irreversible=, strict=, fail_closed=, timeout=.

Google ADK (Runner-wide)

from google.adk.apps import App
from shield_kya.integrations.adk import KyaPlugin

app = App(name="my_app", root_agent=agent, plugins=[KyaPlugin()])
runner = InMemoryRunner(app=app)   # or Runner(app=app, ...)

KyaPlugin is a BasePlugin whose before_tool_callback checks every tool call the Runner dispatches, with the CLI run as an asyncio subprocess (Runner.run included, since ADK runs tools on its event loop). A DENY returns {"error": "KYA denied postgres__drop_table: NEVER_EVENT"}, which ADK uses as the tool result without running the tool, so the model reads the reason and the run continues. ADK re-raises any exception from a plugin as a RuntimeError that aborts the run, so the plugin never lets one escape: a shim error goes through the fail-open / never-list / fail-closed rules. strict=True raises KyaDenied on purpose (ADK wraps it in that RuntimeError) and also blocks REQUIRE_APPROVE.

Covered (one plugin on the App): every agent the Runner runs, including sub_agents reached by transfer_to_agent, an AgentTool with its default include_plugins=True, MCP toolsets and long-running tools (the first call is checked), all tested; live (bidi) mode and workflow tool nodes call the same plugin hook in the google-adk 2.11.0 source (not tested here). ADK's own transfer_to_agent and AgentTool calls are checked too (UNKNOWN_TOOL, advisory; strict=True blocks them). Not covered: AgentTool(include_plugins=False), a separate Runner or App without the plugin, tools the model provider runs server-side (google_search, built-in code execution), an agent code_executor running model-written code (not a tool call), and direct tool.run_async() calls. Plugins run in list order and an agent's own before_tool_callback runs after them, so put KyaPlugin last in plugins and do not rewrite tool args in agent callbacks. Options: server=, irreversible=, strict=, fail_closed=, timeout=, name= (plugin name, default shield_kya).

Microsoft Agent Framework (agent-wide)

from agent_framework import Agent
from shield_kya.integrations.agent_framework import KyaMiddleware

agent = Agent(client=client, tools=[...], middleware=[KyaMiddleware()])

KyaMiddleware is a FunctionMiddleware (package agent-framework-core, the AutoGen successor). Before a function runs it is checked with the CLI as an asyncio subprocess. A DENY sets context.result to the reason and does not call call_next(), so the function never runs, the model receives KYA denied postgres__drop_table: NEVER_EVENT as the function result and the run continues. Agent Framework turns an ordinary exception from middleware into a tool-error result (the tool does not run), so the middleware never raises one: a shim error goes through the fail-open / never-list / fail-closed rules instead of silently blocking a harmless tool, and a tool's own exception stays the tool's. strict=True raises MiddlewareFailure (from KyaDenied), Agent Framework's fail-closed signal that aborts the run, and also blocks REQUIRE_APPROVE. Hosted fail_closed=True / KYA_FAIL_CLOSED=1 blocks the tool the same way as a deny and keeps the run going.

Covered: every function call of that agent's function-invocation loop (run and run(stream=True)), and MCP functions (tested); approved replays of approval_mode="always_require" tools go through the same middleware in the agent-framework-core 1.21.0 source (not tested here). Middleware on the chat client (client = SomeClient(middleware=[KyaMiddleware()])) covers every agent that uses the client. Not covered: an agent wrapped with as_tool() unless that agent (or its client) has its own KyaMiddleware (the as_tool call itself is checked), hosted tools the service runs (code interpreter, web search, hosted MCP), and direct FunctionTool.invoke() calls. The first middleware in the list is the outermost, so put KyaMiddleware last: argument-repair middleware must run before it (Agent Framework's own rule for security middleware). Options: server=, irreversible=, strict=, fail_closed=, timeout=.

OpenAI Agents SDK (every function tool present at attach() time, except SandboxAgent tools added at run time)

from agents import Agent, Runner
from shield_kya.integrations.openai_agents import attach

agent = attach(Agent(name="a", tools=[...], mcp_servers=[...], handoffs=[...]))
Runner.run_sync(agent, "...")

The SDK has no agent-wide pre-tool hook that can skip a tool (RunHooks.on_tool_start can only abort the whole run), only per-tool tool_input_guardrails, so attach() puts one KYA guardrail (named shield_kya) on every FunctionTool in agent.tools and on every MCP server in agent.mcp_servers, and walks handoffs (an Agent or a handoff(agent)) and agent.as_tool() agents recursively. The SDK copies a server's guardrails onto every MCP tool it lists, so MCP tools a server adds later are covered without another call.

Not covered. Shell and file-editing tools: ShellTool, LocalShellTool, ApplyPatchTool, ComputerTool, CustomTool, and the shell, filesystem, apply-patch and image tools a SandboxAgent adds at run time (sandbox capabilities, created after attach()). An agent with them runs commands and edits files without a KYA check; for coding agents the shell and apply-patch tools are the dangerous ones, and this integration does not gate them. Also not covered: hosted tools the OpenAI service runs (WebSearchTool, FileSearchTool, CodeInterpreterTool, ImageGenerationTool, HostedMCPTool, ToolSearchTool), the handoff call itself (only the target agent's tools are checked), the commands Codex runs inside the experimental codex_tool() (the codex call itself is checked), function tools added to agent.tools after attach() (call it again), agents reached only through a custom Handoff without an agent reference, and direct on_invoke_tool calls. ProgrammaticToolCallingTool was not checked.

attach() returns the agent and is idempotent: a second call replaces the KYA guardrail and keeps your own guardrails. For one tool, kya_tool_input_guardrail() returns the guardrail to put in tool_input_guardrails=[...] yourself; for an MCP server use attach([server]) or pass mcp_server_name=server.name, because the guardrail cannot read the server from the call (without it drop_table is checked alone and the never-list cannot tell it is postgres__drop_table).

Before a tool runs, the guardrail checks the call with the CLI as an asyncio subprocess (Runner.run_sync included, since the SDK runs guardrails on its event loop). A DENY returns ToolGuardrailFunctionOutput.reject_content(reason): the tool does not run, the model receives KYA denied postgres__drop_table: NEVER_EVENT as the tool output and the run continues. The SDK aborts the run on any exception from a guardrail, so the guardrail never raises one: a shim error goes through the fail-open / never-list / fail-closed rules. strict=True returns raise_exception instead (the SDK raises ToolInputGuardrailTripwireTriggered and the run aborts) and also blocks REQUIRE_APPROVE. Hosted fail_closed=True / KYA_FAIL_CLOSED=1 blocks the tool the same way as a deny and keeps the run going.

The tool id is the name the model called (ToolContext.tool_name), with the arguments the tool runs with. The never-list is also matched against <namespace>.<name> for a tool_namespace() tool and, for an MCP tool, against <server name>__<tool name>: an MCP server created with name="postgres" exposes drop_table (or mcp_postgres__drop_table with mcp_config={"include_server_in_tool_names": True}), and the check matches it as postgres__drop_table (the deny reason names both). Covered (tested): function tools of the agent, of handoff targets and of as_tool() agents, MCP server tools (both naming modes), namespaced tools, Runner.run, Runner.run_sync and Runner.run_streamed. Realtime agents run the same tool input guardrails in the openai-agents 0.23.1 source (not tested here). With RunConfig(tool_execution=...) pre_approval_tool_input_guardrails=True the guardrail runs before approval and again at execution, so an approved call has two trail rows. Options: server= (function tools only: an MCP tool already carries its server name), irreversible=, strict=, fail_closed=, timeout=.

Tested versions

CI runs the unit tests of all six at the versions above, and the real-CLI e2e (tests/test_e2e_*.py) for Pydantic AI, CrewAI, Google ADK, Agent Framework and the OpenAI Agents SDK; the LangChain e2e is still a manual gate.

Known limits

• KYA_CLI set to an absolute path is trusted as is: the shim runs it. Its verdict cannot ALLOW a never-list tool (the never-list is matched in Python on every call), but the program itself still runs.
• crewai calls load_dotenv() at import. From python -c, a notebook or a debugger it reads the current directory's .env, otherwise the first .env above crewai's install directory (a .venv inside a repo reaches the repo's .env). A cloned repo can therefore set KYA_CLI for a CrewAI process (the never-list still blocks, see above). load_dotenv() does not replace variables that are already set, so exporting KYA_CLI yourself keeps a repo .env from choosing it. python-dotenv also expands ${PWD} and other variables, so a .env can name absolute paths inside the repo. The child's PATH and HOME are only safe because crewai calls load_dotenv() without override=True, so an exported value wins; a .env that sets HOME or PATH for a process where they are unset is a risk the shim does not defend against.
• An empty tool id is denied (EMPTY_TOOL), like the CLI's offline evaluate; kya hook allows a hook payload that has no tool name.
• A grandchild that leaves the CLI's process group and keeps its output open can stretch a timeout past the 4 s budget. The real CLI's detached helpers do not hold its output.
• The child's PATH keeps only absolute entries, so a #!/usr/bin/env node launcher cannot pick up a repo-planted ./node.
• CrewAI hooks are synchronous. Under crew.akickoff() the check blocks the event loop for up to timeout (4 s by default) per tool call; there is no async hook to use instead.
• CrewAI's register() is process-wide: it covers every crew and agent in the process, not one crew.
• The CrewAI reason hand-off from the pre hook to the post hook is per thread. If another pre-tool hook blocks a call before ours runs, CrewAI's own text stays (the tool is still blocked).
• Google ADK: adk run, adk web and adk api_server load the agent folder's .env with load_dotenv(override=True), so under those commands a repo .env can replace PATH, HOME, KYA_CLI and KYA_FAIL_CLOSED (the allowlist above does not help: these are the values it forwards or reads). The never-list still blocks by name, but a planted KYA_CLI or PATH entry runs. In the google-adk 2.11.0 source only those CLI commands load a .env; a program that builds its own Runner does not (other libraries it imports may). agent-framework-core reads a .env only when you pass env_file_path (settings read it without changing os.environ; the observability setup loads it without override).
• The never-list only recognizes MCP tools named server__tool (or server.tool, server/tool, mcp__server__tool) with a known server. Google ADK and Agent Framework keep the original MCP name, so their plugin and middleware also check that name and <prefix>__<original>: an ADK toolset or Agent Framework MCPTool with tool_name_prefix="postgres" exposes postgres_drop_table, and the check matches it as postgres__drop_table (the deny reason names both). Without a prefix, an MCP tool is just drop_table in both frameworks and the never-list cannot tell which server it belongs to (server= applies one prefix to every tool). Agent Framework replaces characters outside A-Za-z0-9_.- with - in the exposed name; the raw remote name is checked as well. A prefix that names a known server can also block a harmless tool whose name contains a never-list verb (github_transfer_issue is checked as github__transfer_issue), the same over-block the never-list has for server__tool names. The other frameworks' own MCP names do not match: Pydantic AI load_mcp_servers() / toolset.prefixed("postgres") gives postgres_drop_table (one _), and CrewAI mcps= names tools <server_name>_<tool>, where a stdio server_name is the command line and an HTTP one the host, so CrewAI MCP tools are never matched. Those calls are still evaluated and recorded, but run (UNKNOWN_TOOL is advisory). What to do: name MCP tools in the server__tool form (Pydantic AI toolset.prefixed("postgres_") gives postgres__drop_table; LangChain server=), or rely on hosted enforcement.
• CrewAI MCP tool ids are the original tool names, which for stdio servers embed the full command line, including any DSN password (postgresql://user:pass@...). That id goes on the kya argv (visible in ps), into ~/.kya/trail.jsonl and into the deny reason. Keep secrets out of MCP command lines (use the server's env instead) when you use register().
• Pydantic AI output tools are not checked.
• Agent Framework: KyaMiddleware on the chat client always runs outside the agent's own function middleware (agent-framework-core 1.21.0 puts client middleware first), so agent-level middleware can rewrite the arguments after the check. With KyaMiddleware on the client, put argument-rewriting function middleware on the client too, before KyaMiddleware, not on the agent.
• server= on the agent-wide Google ADK, Agent Framework and OpenAI Agents SDK hooks prefixes the tool id sent to the CLI, but the never-list also matches every name without that prefix, so a prefix it does not know (server="acme") cannot hide postgres__drop_table; the denied call is recorded under the unprefixed id.
• Google ADK: the original MCP name is read from McpTool.raw_mcp_tool (public in google-adk 2.11.0), else the private _mcp_tool. The tests use stand-in objects with one or the other attribute, not a real McpTool.
• Google ADK: ManagedAgent tools and RemoteMcpServer tools run on Google's backend (ADK never connects to that MCP server), so they never reach before_tool_callback and are not checked.
• OpenAI Agents SDK: the guardrail is per tool, so a function tool added to agent.tools after attach() is not checked until attach() runs again (MCP tools are, through the server). An MCP tool name the SDK shortened with a hash (server plus tool name over 64 characters) is not matched by its <server>__<tool> form. An MCP server name or a tool_namespace() name that is a known server can also block a harmless tool whose name contains a never-list verb (server github with transfer_issue is checked as github__transfer_issue), the same over-block as above. Options live on each tool and server object: when two agents share one, the last attach() wins for both (attach(a, fail_closed=True) then attach(b) makes a shared tool fail open for a too). Shell and file-editing tools (ShellTool, LocalShellTool, ApplyPatchTool, ComputerTool, CustomTool, SandboxAgent run-time tools) are not checked; see Not covered in the OpenAI Agents SDK section.
• The CLI child gets an allowlist of the environment, nothing else: an absolute-only PATH, HOME, USERPROFILE, LANG, LC_*, TZ, on Windows also SYSTEMROOT, WINDIR, COMSPEC, PATHEXT, APPDATA, LOCALAPPDATA, TEMP and TMP, plus KYA_SKIP_NODE_CHECK=1 set by the shim (TMPDIR is left out). A framework can load a repo .env into os.environ (crewai does at import), and other variables would run repo code in node (NODE_OPTIONS=--require ./x.js, OPENSSL_CONF loading a provider library, LD_PRELOAD), move the trail (KYA_HOME) or send trail events to a webhook (KYA_NOTIFY_WEBHOOK, KYA_OTLP_EXPORT_ENDPOINT). A user-exported value looks the same, so CLI settings such as KYA_HOME, KYA_SESSION_ID, KYA_TELEMETRY, KYA_NOTIFY_WEBHOOK, proxy and CA variables are not supported through the shim: the trail always goes to $HOME/.kya. The privacy opt-outs KYA_DIFF_PREVIEW and KYA_SEND_PREVIEWS are forwarded only when set to an off value (0, false, no, off), so a planted value can only remove data. KYA_CLI and KYA_FAIL_CLOSED are read by the shim itself and still work.
• A committed repo .kya/config.json still applies: the CLI runs in the agent's working directory and reads that file, so a repo can add notify webhooks or an OTLP endpoint that receive trail events (every DENY and REQUIRE_APPROVE, with tool ids and summaries). This is CLI behaviour, the same for kya hook, not something the shim's environment allowlist can stop. CLI follow-up: do not honor repo-level webhook config in hook mode.
• CrewAI: when two tools sanitize to the same name (postgres__drop_table and postgres_drop_table), CrewAI can hand the hook one and run the other, so the hook denies the call if any tool with that sanitized name on the agent, the task, the agent's executor (mcps=, apps=, execute_task(tools=)) or a LiteAgent is on the never-list. The harmless twin is then blocked too, with a reason that names both: KYA denied postgres_drop_table: a tool with the same sanitized name (postgres__drop_table) is on the never-list (NEVER_EVENT); the trail row is the DENY for postgres__drop_table.
• CrewAI strict=True also blocks CrewAI's own delegation tools (delegate_work_to_coworker, ask_question_to_coworker), which evaluate as REQUIRE_APPROVE UNKNOWN_TOOL.
• The legacy per-tool wrappers (governed_tool, wrap_tool, wrap_tools) still raise KyaDenied on a deny, which aborts a Pydantic AI run; prefer the agent-wide entry points.

Core API

import shield_kya

d = shield_kya.decide("web_search", {"q": "kya"})       # never raises; records the trail
d.allowed, d.verdict, d.reason_code, d.reason
d = await shield_kya.adecide("web_search", {"q": "kya"})  # async, off the event loop

shield_kya.evaluate("org.sample.data.write", args={"table": "users"})  # raw eval-tool dict; raises on CLI errors

@governed gates a sync or async callable. A plain function has no other way to report a deny, so DENY raises KyaDenied before the function runs:

from shield_kya import governed, KyaDenied

@governed("org.sample.data.write", irreversible=True)
def write_users_table(rows): ...

try:
    write_users_table(rows)
except KyaDenied as denied:
    print(denied.verdict, denied.reason_code)

Per-tool wrappers

Prefer the agent-wide entry points above. These wrap single tools and share the governed semantics (DENY raises KyaDenied). Framework imports are lazy.

from shield_kya.integrations.langgraph import governed_tool   # LangGraph / LangChain @tool
tools = [governed_tool("web.fetch")(fetch_fn)]

from shield_kya.integrations.crewai import wrap_tool          # CrewAI BaseTool or callable
safe_search = wrap_tool(search_tool, tool_id="search.web")

from shield_kya.integrations.pydantic_ai import governed_tool  # Pydantic AI

@agent.tool_plain
@governed_tool("search.web")
def search_web(q: str) -> str: ...

A raised KyaDenied aborts a Pydantic AI run (Pydantic AI does not turn tool exceptions into tool errors).

Extras (pip install '.[name]' from a checkout): langchain, langgraph, crewai (>=1.15.27,<2), pydantic-ai (pydantic-ai-slim >=2.55.0,<3), adk (google-adk >=2.11.0,<3), agent-framework (agent-framework-core >=1.21.0,<2), openai-agents (openai-agents >=0.23.1,<0.24).

Tests

python -m unittest discover -s tests   # stdlib suite; framework tests run when the framework is installed
KYA_E2E_BIN=/dir/with/kya python -m unittest discover -s tests -p 'test_e2e_*.py'   # real-CLI e2e
KYA_E2E_BIN=/dir/with/kya python -m unittest discover -s tests -p 'test_shield_kya.py' -k RealCliEnvTest   # real CLI: a planted OPENSSL_CONF never reaches node (POSIX, needs cc for the full check)

License

MIT