ai-control-plane 2.0.1

AI Control Plane Python SDK — observability, cost tracking, and analytics for LLM applications

AI Control Plane — Python SDK

Operational infrastructure for production AI agents. Add observability, cost tracking, and usage analytics to any LLM application with 2 lines of code.

Installation

pip install ai-control-plane

With provider extras:

pip install ai-control-plane[openai]      # OpenAI support
pip install ai-control-plane[anthropic]   # Anthropic support
pip install ai-control-plane[all]         # All providers

Quick Start

from openai import OpenAI
from acp_sdk import wrap

# Wrap your existing client — that's it
client = wrap(OpenAI(), api_key="acp_live_...", app="my-app")

# Use exactly as before — same interface, zero code changes
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)

Every call is now automatically tracked: tokens, cost, latency, and status — shipped asynchronously so your application is never slowed down.

Features

• 2-line integration — import + wrap(), nothing else changes
• Zero latency overhead — telemetry ships asynchronously in a background thread
• Multi-provider — OpenAI, Anthropic, and Google clients supported
• Automatic cost tracking — built-in pricing for GPT-4o, Claude 3.5, and more
• Rich metadata — attach user IDs, session IDs, agent IDs, and custom tags
• Fail-open design — if telemetry fails, your LLM calls still work
• Type-safe config — ACPConfig dataclass with environment variable support

Usage

OpenAI

from openai import OpenAI
from acp_sdk import wrap

client = wrap(OpenAI(), api_key="acp_live_...", app="my-app")

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    _acp_user_id="user_123",
    _acp_session_id="session_abc",
)

Anthropic

import anthropic
from acp_sdk import wrap

client = wrap(anthropic.Anthropic(), api_key="acp_live_...", app="my-app")

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    _acp_user_id="user_123",
)

Embeddings

from openai import OpenAI
from acp_sdk import wrap

client = wrap(OpenAI(), api_key="acp_live_...", app="my-app")

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="The quick brown fox",
)

ACP Metadata Parameters

Attach metadata to any call using _acp_ prefixed parameters:

Parameter	Description
_acp_user_id	End-user identifier for per-user analytics
_acp_session_id	Session/conversation identifier
_acp_agent_id	Agent identifier (for multi-agent systems)
_acp_run_id	Run/execution identifier
_acp_tags	Custom tags dict for filtering and grouping

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello"}],
    _acp_user_id="user_123",
    _acp_session_id="sess_abc",
    _acp_agent_id="agent_research",
    _acp_run_id="run_456",
    _acp_tags={"team": "backend", "feature": "search"},
)

Configuration

from acp_sdk import wrap, ACPConfig

# Option 1: Inline parameters
client = wrap(OpenAI(), api_key="acp_live_...", app="my-app", environment="staging")

# Option 2: Config object
config = ACPConfig(
    api_key="acp_live_...",
    app="my-app",
    environment="production",
    tags={"team": "ml-platform"},
)
client = wrap(OpenAI(), config=config)

# Option 3: Environment variables
# Set ACP_API_KEY, ACP_APP, ACP_ENV, ACP_INGEST_URL
config = ACPConfig.from_env()
client = wrap(OpenAI(), config=config)

Flush Before Exit

The SDK buffers telemetry and sends it in batches. Call flush() before your process exits to ensure all events are delivered:

client.flush(timeout=5.0)

What Gets Tracked

Every LLM call automatically captures:

Field	Description
provider	openai, anthropic, or google
model_id	Model name (gpt-4o, claude-3-5-sonnet, etc.)
input_tokens	Prompt/input token count
output_tokens	Completion/output token count
cost_usd	Estimated cost in USD
latency_ms	End-to-end call duration
status	success or error
error_message	Error details (on failure)

Requirements

• Python 3.10+
• httpx >= 0.27.0
• pydantic >= 2.0

Links

• Dashboard: ai-control-plane.in
• Source: github.com/Anil175/ai-control-plane

License

MIT