Python SDK

pip install agent-control-sdk

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}"

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",
)

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

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

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",
    )

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

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())

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:

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

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

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

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)

pip install agent-control-sdk

import agent_control

import logging

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

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

from agent_control.settings import configure_settings

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

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

from agent_control import get_logger

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

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