Agent Control - Python SDK
Python SDK for adding safety controls and guardrails to AI agents. Protect tools and LLM calls with server-side or local policy enforcement.
Installation
pip install agent-control-sdk
Quick Start
import agent_control
from agent_control import control
# Initialize
agent_control.init(agent_name="my-agent")
# Protect a tool with decorator
@control()
async def search_database(query: str) -> str:
return db.execute(query)
Controls are defined centrally and enforced automatically at runtime. See Python SDK Documentation for complete reference.
Supplying available tools to an LLM control
The @control decorator is a convenience path for function-based steps. An LLM
control can include static tool definitions, or a provider can build them from
the function arguments for each pre- and post-execution check:
AVAILABLE_TOOLS = [
{
"type": "function",
"function": {
"name": "search",
"description": "Search internal documents",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
}
]
@agent_control.control(step_type="llm", tools=AVAILABLE_TOOLS)
async def answer(message: str) -> str:
...
For per-call tools, pass a provider such as
tools=lambda args, kwargs: tools_for(kwargs.get("model")). The provider
receives the raw positional call arguments in args and keyword arguments in
kwargs for each pre- and post-execution check. The provider may return None
when the step has no available tools.
For framework adapters and advanced callers, construct the shared Step
directly and send it through the SDK's check_evaluation helper. This keeps
framework-specific message and tool formats in the adapter, without requiring
the decorator or a framework dependency:
step = agent_control.Step(
type="llm",
name="openai_agent",
input=messages,
output=response,
tools=available_tools,
)
async with agent_control.AgentControlClient() as client:
result = await agent_control.evaluation.check_evaluation(
client,
agent_name="my-agent",
step=step,
stage="post",
)
The existing evaluate_controls helper remains available for callers that
prefer its field-based convenience arguments.
Building a step with children
A control's scope can target any step type, including one that aggregates
several already-executed child steps - Step.children just has to be
populated by the caller. Building that tree by hand means hand-rolling a
side-channel record for every leaf call and converting it into Step
objects afterward:
# Before: a hand-built dict record next to the real return value
def execute_policy_search(query):
docs = search_policy_documents(query)
record = {"type": "lookup", "query": query, "docs": docs}
return StepExecution(value=docs, record=record)
children = []
result = execute_policy_search(query)
children.append(result.record)
...
workflow_step = build_agent_control_step({"type": "workflow", "children": children, ...})
agent_control.record_step() builds the same tree incrementally, in Step
vocabulary, with no intermediate dict and no converter:
with agent_control.record_step("workflow", "banking_workflow", input={"request": req}) as workflow:
with workflow.child("lookup", "policy_lookup", input=query) as span:
span.output = search_policy_documents(query)
account = workflow.call(lookup_account, account_id="acct-1001", step_type="tool")
plan = workflow.call(run_banking_model, req, step_type="llm", tools=TOOL_DEFINITIONS)
workflow.output = {"status": "planned", "message": plan["content"]}
result = await workflow.evaluate(stage="post")
workflow.call(...)/await workflow.acall(...) run the function and record
it as a child using the same capture logic as @control() (input from bound
arguments, output from the return value), then return the real result -
removing the need for a separate StepExecution-style wrapper. A failed
call is still recorded (with the error in context) before the exception is
re-raised. workflow.child(...) nests another recorder the same way, so a
parent step can nest children of any depth, e.g. session.child("trace", ...) for callers that model multi-turn sessions of traces.
workflow.build() produces the frozen Step, with children=[] instead of
omitted (None) for a step type passed via container_types even when no
children were recorded - e.g. record_step("workflow", ..., container_types= {"workflow"}) for a workflow that legitimately ran with zero children.
workflow.evaluate(...) builds the step and evaluates it in one call via
evaluate_step() - the same function evaluate_controls() uses internally
once its Step is built.
Sharing an OpenTelemetry provider with Google ADK
When Google ADK and Agent Control should export through the same OpenTelemetry
pipeline, configure one SDK TracerProvider and pass it to both components:
import agent_control
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from openinference.instrumentation.google_adk import GoogleADKInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
resource = Resource.create({"service.name": "my-google-adk-agent"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
GoogleADKInstrumentor().instrument(tracer_provider=provider)
agent_control.init(
agent_name="my-google-adk-agent",
observability_enabled=True,
observability_sink_name="otel",
observability_sink_config={"enabled": True},
otel_tracer_provider=provider,
)
Agent Control reuses this provider without adding an exporter or span processor. It force-flushes the provider during shutdown but leaves provider shutdown to the application. If no provider is passed, the OTEL sink first checks for a globally registered SDK provider, then falls back to its existing Agent Control-owned OTLP pipeline.
1 maintainer
Metadata
1 maintainer
Release files for agent-control-sdk 8.14.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| agent_control_sdk-8.14.0.tar.gz | 223.3 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| agent_control_sdk-8.14.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 366.8 kB
Release files / agent_control_sdk-8.14.0.tar.gz
| Download URL | agent_control_sdk-8.14.0.tar.gz |
|---|---|
| Size | 223.3 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
9f10237194babcc9a37f9cf2c80de6dfd4dfd95f921f9ba9a731eed9c33bfbba
|
|
BLAKE2b-256 checksum How to use checksums |
7d7229ef689cba9c89f87da76fa0208b91950b6d36ea29e68fb858f522f9575c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / agent_control_sdk-8.14.0-py3-none-any.whl
| Download URL | agent_control_sdk-8.14.0-py3-none-any.whl |
|---|---|
| Size | 143.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
6600326a467cf3f2eeb75f54ba54954480aa7203906442f7ba1266fdef97e4f6
|
|
BLAKE2b-256 checksum How to use checksums |
b85f8151d3ad03ef4095b2bad5eb86ac1d8882864edd03e8d349adecc6e02157
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Security for agent-control-sdk 8.14.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
