agentshield-control-plane 0.1.0

AgentShield SDK — instrument AI agents and capture execution events for the AgentShield Runtime Control Plane

agentshield-control-plane

The Python SDK for AgentShield — instrument any AI agent and stream what it actually does (tool calls, LLM outputs, successes, failures) to an AgentShield backend, where every action becomes an immutable, classified, queryable audit event.

Framework-agnostic by design: works with LangChain, CrewAI, and the OpenAI Agents SDK out of the box, and with any other Python agent code via three manual calls.

Install

pip install agentshield-control-plane

# with a framework adapter
pip install "agentshield-control-plane[langchain]"
pip install "agentshield-control-plane[crewai]"
pip install "agentshield-control-plane[openai-agents]"

Requires Python 3.10+.

Before you start: you need an AgentShield backend

The SDK only ever does one thing: turn real execution into events and send them somewhere. It does not include a backend. You need:

1. A running AgentShield backend instance (self-hosted — see backend/README.md in the AgentShield monorepo — or one provided to you directly if you're a design partner).
2. A tenant API key for that instance, issued server-side (scripts/issue_api_key.py).

There is no public hosted endpoint yet — shield = AgentShield(api_key=...) with no endpoint argument will try https://api.agentshield.dev, which does not exist today. Pass your own backend's URL explicitly.

Quickstart (manual instrumentation)

Works with any agent code, no framework required:

from agentshield_control_plane import AgentShield

shield = AgentShield(api_key="ags_live_xxxx", endpoint="http://localhost:8000")

with shield.session(agent="my-agent") as session:
    with session.tool_call("search", query="weather in SF"):
        ...  # your real tool-call code
    with session.llm_call("gpt-5", tokens=120):
        ...  # your real LLM call

shield.shutdown()  # optional — atexit flushes best-effort if you skip this

That's it. Within a couple seconds, GET /api/v1/sessions/{session_id} on your backend returns this session and its events.

Framework adapters

Adapters hook each framework's own callback/event system — they never monkey-patch internals, so your agent's behavior is untouched.

LangChain

from agentshield_control_plane import AgentShield
from agentshield_control_plane.adapters.langchain import AgentGuardCallbackHandler

shield = AgentShield(api_key="ags_live_xxxx", endpoint="http://localhost:8000")
handler = AgentGuardCallbackHandler(shield, agent_id="my-agent")

chain.invoke(input, config={"callbacks": [handler]})

One handler instance tracks any number of concurrent top-level chain invocations — each root-level invoke() (no parent_run_id) opens its own session; nested tool/LLM calls are attributed back to it via LangChain's own run_id/parent_run_id chain.

CrewAI

from agentshield_control_plane import AgentShield
from agentshield_control_plane.adapters.crewai import AgentGuardEventListener

shield = AgentShield(api_key="ags_live_xxxx", endpoint="http://localhost:8000")
AgentGuardEventListener(shield, agent_id="my-agent")  # registers globally — instantiate once

crew.kickoff()

CrewAI's event bus is a process-wide singleton, so AgentGuardEventListener is instantiated once (not passed per-call like the LangChain handler) and listens for the lifetime of the process. It opens a session on crew.kickoff() and tracks tool usage against whichever session is currently open — correct for the common case of one kickoff at a time per process. A process running genuinely concurrent, overlapping kickoff() calls on the same listener instance will not separate their tool calls correctly; this is a deliberate, named simplification, not an oversight (see the adapter's docstring).

OpenAI Agents SDK

from agentshield_control_plane import AgentShield
from agentshield_control_plane.adapters.openai_agents import AgentGuardTracingProcessor
from agents import add_trace_processor

shield = AgentShield(api_key="ags_live_xxxx", endpoint="http://localhost:8000")
add_trace_processor(AgentGuardTracingProcessor(shield, agent_id="my-agent"))  # once, globally

await Runner.run(agent, "...")

Hooks the SDK's own tracing system (TracingProcessor/add_trace_processor) — additive, so it runs alongside whatever else you already have registered, including the SDK's own default export. Like the CrewAI listener, register one instance once, globally; unlike it, this adapter correctly tracks multiple concurrent runs, because each run's spans carry a real trace_id back-reference. Tool calls and model calls both map to real AgentShield events; a handful of internal span types (agent/task/turn/handoff) don't have a canonical event slot yet and aren't reported — the same "no slot, don't report" choice already made for LangChain LLM failures, not a new gap. One real, verified limitation: this adapter can't currently tell whether the overall run failed — only individual tool/handoff failures are visible to it — so agent.execution.finished is always reported as success. A real exception from Runner.run() still propagates to your own code exactly as it would without this adapter; only the AgentShield-side telemetry can't reflect that specific case yet.

Configuration

Argument / env var	Default	What it does
api_key / AGENTSHIELD_API_KEY	(required)	Your tenant's API key
endpoint / AGENTSHIELD_ENDPOINT	https://api.agentshield.dev	Backend base URL — see the warning above
environment / AGENTSHIELD_ENVIRONMENT	production	Tagged onto events for your own filtering
capture_prompts	True	Include prompt text in llm_call events
capture_outputs	True	Include output text in tool_call/llm_call events
batch_size	100	Events per batch sent to the backend
flush_interval	2.0	Max seconds an event waits in the queue before a flush

Everything not resolved via an explicit constructor argument falls back to its environment variable, then its default. AgentShield() raises ConfigurationError immediately if no API key is found anywhere.

Architecture

agentshield_control_plane/
├── core/          AgentShield (entry point), Session (manual instrumentation), lifecycle
├── models/        SDK-owned event types and the RawEvent shape — no backend imports
├── serialization/ RawEvent -> wire-format dict, with a redaction-hook seam
├── queue/         bounded buffer, batcher (size/time flush), background sender thread
├── transport/     Transport protocol; HTTPTransport is the only implementation today
└── adapters/      langchain.py, crewai.py, openai_agents.py — hook each framework's own
                   event/callback/tracing system

• The SDK owns its own wire-format models. It does not import anything from the backend — it's a separately published package. EventType values in models/events.py are a deliberate, manually-synced subset of the backend's canonical event catalog.
• The serializer is a privacy boundary. Every event crosses EventSerializer.serialize() exactly once before going anywhere. capture_prompts/capture_outputs are built in there; a redactors hook list is where custom PII detection/masking plugs in.
• Graceful shutdown. AgentShield.shutdown() stops the background sender and flushes whatever's queued (bounded by shutdown_flush_timeout, default 5s). An atexit hook runs the same logic best-effort if you never call it explicitly.
• This is an event-producing runtime shim, not an Agent CRUD client. Nothing in this package registers, creates, or looks up an Agent via any API — agent_id/session_id are just fields on the events it sends.

Security & Data FAQ

AgentShield is a security and compliance product, so an open SDK is a trust advantage, not just a developer-experience nicety: every claim below can be checked directly against the source in this repository rather than taken on faith.

What exactly does the SDK collect from my agents? Only what you explicitly instrument: tool-call name and arguments/result, LLM call model name and token counts, success/failure status, and timing — wrapped into the event types in src/agentshield_control_plane/models/events.py. Nothing is collected by inspecting your process, your filesystem, or any code path you didn't wrap in a session.tool_call(...)/session.llm_call(...) (or the equivalent framework adapter hook). There is no background scanning, no auto-instrumentation of unrelated code, and no telemetry collection beyond the events you generate.

Where does that data go? Exactly one place: the endpoint you configure, via HTTPTransport — the only transport this SDK implements (src/agentshield_control_plane/transport/). There is no AgentShield-operated collection service it reports to in the background, and no public hosted endpoint exists today (see "Before you start," above). If you self-host the backend, your event data never leaves your own infrastructure.

Can I verify this myself instead of taking your word for it? Yes — that's the point of shipping this as a real, MIT-licensed, source-available package rather than a compiled or obfuscated agent. Every event this SDK can produce is enumerated in src/agentshield_control_plane/models/events.py; every byte that leaves your process passes through EventSerializer.serialize() in src/agentshield_control_plane/serialization/ exactly once, and that function is the entire surface you need to read to know what's sent. There's no second, hidden code path.

Can I keep prompt or output text out of what's sent? Yes — capture_prompts and capture_outputs (both default True, see Configuration above) control whether prompt/output text is included at all. Setting either to False removes that content from the corresponding events entirely, not just from a display layer. A redactors hook list in the serializer is the integration point for your own PII detection/masking before anything is serialized, if you need partial redaction rather than an all-or-nothing toggle.

Does the SDK phone home to AgentShield, the company, separately from my own backend? No. There is exactly one network destination: the endpoint you configure. If that's a self-hosted backend you control, AgentShield (the company) never sees your data at all.

Does the SDK store anything on disk? No. The internal queue (src/agentshield_control_plane/queue/buffer.py) is a plain in-process queue.Queue — nothing is written to disk at any point. The tradeoff, named directly in "Not yet built" below: if your process exits while events are still queued and unsent, they're lost, not persisted for later. There's no local cache, log file, or database file this SDK creates anywhere on the filesystem.

Not yet built

• Auto-detection of installed frameworks (AgentShield.instrument() exists as a stable no-op for forward compatibility, but does nothing yet — wire up the adapter you need explicitly, as shown above).
• Disk-backed offline buffering (in-memory only — events queued during a backend outage are lost if the process exits before it recovers).
• PII detection/masking beyond the capture_prompts/capture_outputs toggles.

Local development

uv sync --all-extras
uv run pytest                              # no network required — see tests/conftest.py
uv run ruff check . && uv run ruff format .
uv run mypy src tests

Manual end-to-end check against a running backend

cd ../backend && docker compose up -d      # if not already running
cd ../sdk
uv run python -c "
from agentshield_control_plane import AgentShield
shield = AgentShield(api_key='<your real key>', endpoint='http://localhost:8000')
with shield.session(agent='manual-test') as s:
    with s.tool_call('ping'): pass
shield.shutdown()
"
curl http://localhost:8000/api/v1/events?agent_id=manual-test \
  -H 'Authorization: Bearer <your real key>'

License

MIT — the LICENSE file ships in this package (both the wheel and sdist) and is also in the GitHub repository.