Python SDK

Unified Python SDK for Agent Control with protection, monitoring, and rule enforcement.

Unified Python SDK for Agent Control, providing agent protection, monitoring, and rule enforcement in one package.

Installation

pip install agent-control-sdk

Quick Start

Simple Initialization

import agent_control

agent_control.init(
    agent_name="customer-service-bot",
    agent_description="My Customer Service Bot",
)

from agent_control import control

@control()
async def handle_message(message: str):
    return f"Processed: {message}"

With Full Metadata

import agent_control

agent_control.init(
    agent_name="customer-service-bot",
    agent_description="Handles customer inquiries and support",
    agent_version="2.1.0",
    server_url="http://localhost:8000",
    team="customer-success",
    environment="production",
)

Features

1. Simple Initialization

One call sets up your agent:

agent_control.init(agent_name="customer-service-bot")

This automatically:

• Creates an Agent instance with your metadata
• Registers with the Agent Control server
• Fetches controls from the server
• Auto-discovers and loads local controls.yaml as a fallback
• Enables the @control() decorator

2. Decorator-Based Protection

Protect any function with server-defined controls:

@control()
async def process(user_text: str):
    return user_text

3. HTTP Client

Use the client directly for custom workflows:

from agent_control import AgentControlClient
import agent_control

async with AgentControlClient() as client:
    health = await client.health_check()
    print(f"Server status: {health['status']}")

    result = await agent_control.evaluation.check_evaluation(
        client,
        agent_name="customer-service-bot",
        step={"type": "llm", "input": "User input here"},
        stage="pre",
    )

4. Agent Metadata

Access your agent information:

agent = agent_control.current_agent()
print(f"Agent: {agent.agent_name}")
print(f"Version: {agent.agent_version}")

Complete Example

import asyncio
import agent_control
from agent_control import control, ControlViolationError

agent_control.init(
    agent_name="customer-support-bot",
    agent_description="Customer Support Bot",
    agent_version="1.0.0",
)

@control()
async def handle_message(message: str) -> str:
    return f"Processed: {message}"

@control()
async def generate_response(query: str) -> str:
    return f"Response with SSN: 123-45-6789"

async def main():
    try:
        result1 = await handle_message("Hello, I need help")
        print(result1)

        result2 = await generate_response("Get user info")
        print(result2)
    except ControlViolationError as e:
        print(f"Blocked by control '{e.control_name}': {e.message}")

asyncio.run(main())

API Reference

Initialization

agent_control.init()

def init(
    agent_name: str,
    agent_description: str | None = None,
    agent_version: str | None = None,
    server_url: str | None = None,
    api_key: str | None = None,
    controls_file: str | None = None,
    steps: list[dict] | None = None,
    conflict_mode: Literal["strict", "overwrite"] = "overwrite",
    observability_enabled: bool | None = None,
    log_config: dict[str, Any] | None = None,
    policy_refresh_interval_seconds: int = 60,
    **kwargs: object,
) -> Agent:

Parameters:

• agent_name: Unique identifier for the agent (normalized to lowercase)
• agent_description: Optional description
• agent_version: Optional version string
• server_url: Optional server URL (defaults to AGENT_CONTROL_URL)
• api_key: Optional API key (defaults to AGENT_CONTROL_API_KEY)
• controls_file: Optional explicit path to controls.yaml
• steps: Optional list of step schemas to register
• conflict_mode: Init registration conflict mode (strict or overwrite)
• observability_enabled: Optional observability toggle
• log_config: SDK logging config
• policy_refresh_interval_seconds: Background controls refresh interval; set to 0 to disable
• **kwargs: Additional metadata

When background refresh is enabled, the SDK refreshes control snapshots via GET /agents/{agent_name}/controls. On refresh failures, it keeps the previous snapshot (fail-open behavior).

Decorator

@control()

def control(policy: str | None = None, step_name: str | None = None):

Decorator to protect a function with server-defined controls.

Parameters:

• policy: Optional policy label for code readability when multiple policies exist
• step_name: Optional step name override for the registered step

Example:

@control()
async def my_func(text: str):
    return text

Exceptions

• ControlViolationError for deny actions
• ControlSteerError for steer actions with a steering context

Client

AgentControlClient

class AgentControlClient:
    def __init__(
        self,
        base_url: str = "http://localhost:8000",
        api_key: str | None = None,
        timeout: float = 30.0,
    ):

Methods:

• health_check()
• Use with module functions like agent_control.agents.*, agent_control.controls.*, etc.

Example:

from agent_control import AgentControlClient
import agent_control

async with AgentControlClient(base_url="http://server") as client:
    health = await client.health_check()
    agent = await agent_control.agents.init_agent(client, agent_data, tools)

Models

If agent-control-models is installed, these classes are available:

• Agent
• ProtectionRequest
• ProtectionResult
• HealthResponse

Configuration

Environment Variables

• AGENT_CONTROL_URL (default: http://localhost:8000)
• AGENT_CONTROL_API_KEY (optional)

Server-Defined Controls

Controls are defined on the server via the API or web dashboard, not in code. This keeps security policies centrally managed and allows updating controls without redeploying your application.

Package Name

This package is named agent-control-sdk (PyPI) and imported as agent_control:

pip install agent-control-sdk

import agent_control

SDK Logging

The SDK uses Python’s standard logging module with loggers under the agent_control.* namespace.

Configuring SDK Logs

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
)

logging.getLogger('agent_control').setLevel(logging.DEBUG)

SDK Settings (Log Categories)

from agent_control.settings import configure_settings

configure_settings(
    log_enabled=True,
    log_span_start=True,
    log_span_end=True,
    log_control_eval=True,
)

Logging Environment Variables

export AGENT_CONTROL_LOG_ENABLED=true
export AGENT_CONTROL_LOG_SPAN_START=true
export AGENT_CONTROL_LOG_SPAN_END=true
export AGENT_CONTROL_LOG_CONTROL_EVAL=true

Using SDK Loggers

from agent_control import get_logger

logger = get_logger(__name__)
logger.info("Processing started")

Default Settings:

• log_enabled: true
• All behavioral settings enabled

Development

cd sdks/python
uv sync
uv run pytest
uv run ruff check .

Examples

• Customer Support Agent
• LangChain SQL Agent
• Galileo Luna-2 Integration

Documentation

• Reference Guide
• Examples Overview
• Architecture (SDK design patterns)