adcp 3.12.0

Official Python client for the Ad Context Protocol (AdCP)

adcp - Python Client for Ad Context Protocol

Official Python SDK for the Ad Context Protocol (AdCP). Build and connect to advertising agents that work synchronously OR asynchronously with the same code.

Building an AdCP Agent

The fastest path to a working agent: subclass ADCPHandler, use response builders, call serve().

from adcp.server import ADCPHandler, serve
from adcp.server.responses import capabilities_response, products_response

class MySeller(ADCPHandler):
    async def get_adcp_capabilities(self, params, context=None):
        return capabilities_response(["media_buy"])

    async def get_products(self, params, context=None):
        return products_response(MY_PRODUCTS)

    # implement create_media_buy, get_media_buys, sync_creatives, etc.

serve(MySeller(), name="my-seller")

Validate with storyboards:

python agent.py &
npx @adcp/client storyboard run http://localhost:3001/mcp media_buy_seller --json

Agent type	Skill	Storyboard	Steps
Seller (publisher, SSP, retail media)	skills/build-seller-agent/	media_buy_seller	9
Signals (audience data, CDP)	skills/build-signals-agent/	signal_owned	4
Creative (ad server, CMP)	skills/build-creative-agent/	creative_lifecycle	6

For compliance testing, add a TestControllerStore so storyboards can force state transitions:

from adcp.server.test_controller import TestControllerStore
serve(MySeller(), name="my-seller", test_controller=MyStore())

Each skill file in skills/ contains the complete pattern, response shapes, and validation loop for coding agents (Claude, Codex) to generate passing servers.

Connecting to AdCP Agents

The Core Concept

AdCP operations are distributed and asynchronous by default. An agent might:

• Complete your request immediately (synchronous)
• Need time to process and send results via webhook (asynchronous)
• Ask for clarifications before proceeding
• Send periodic status updates as work progresses

Your code stays the same. You write handlers once, and they work for both sync completions and webhook deliveries.

Installation

pip install adcp

Note: This client requires Python 3.10 or later and supports both synchronous and asynchronous workflows.

Quick Start: Test Helpers

The fastest way to get started is using pre-configured test agents with the .simple API:

from adcp.testing import test_agent

# Zero configuration - just import and call with kwargs!
products = await test_agent.simple.get_products(
    brief='Coffee subscription service for busy professionals',
    buying_mode='brief',
)

print(f"Found {len(products.products)} products")

Simple vs. Standard API

Every ADCPClient includes both API styles via the .simple accessor:

Simple API (client.simple.*) - Recommended for examples/prototyping:

from adcp.testing import test_agent

# Kwargs and direct return - raises on error
products = await test_agent.simple.get_products(brief='Coffee brands', buying_mode='brief')
print(products.products[0].name)

Standard API (client.*) - Recommended for production:

from adcp.testing import test_agent
from adcp import GetProductsRequest

# Explicit request objects and TaskResult wrapper
request = GetProductsRequest(brief='Coffee brands', buying_mode='brief')
result = await test_agent.get_products(request)

if result.success and result.data:
    print(result.data.products[0].name)
else:
    print(f"Error: {result.error}")

When to use which:

• Simple API (.simple): Quick testing, documentation, examples, notebooks
• Standard API: Production code, complex error handling, webhook workflows

Available Test Helpers

Pre-configured agents (all include .simple accessor):

• test_agent: MCP test agent with authentication
• test_agent_a2a: A2A test agent with authentication
• test_agent_no_auth: MCP test agent without authentication
• test_agent_a2a_no_auth: A2A test agent without authentication
• creative_agent: Reference creative agent for preview functionality
• test_agent_client: Multi-agent client with both protocols

Note: Test agents are rate-limited and for testing/examples only. DO NOT use in production.

See examples/simple_api_demo.py for a complete comparison.

Tip: Import types from the main adcp package (e.g., from adcp import GetProductsRequest) rather than adcp.types.generated for better API stability.

Quick Start: Distributed Operations

For production use, configure your own agents:

from adcp import ADCPMultiAgentClient, AgentConfig, GetProductsRequest

# Configure agents and handlers (context manager ensures proper cleanup)
async with ADCPMultiAgentClient(
    agents=[
        AgentConfig(
            id="agent_x",
            agent_uri="https://agent-x.com",
            protocol="a2a"
        ),
        AgentConfig(
            id="agent_y",
            agent_uri="https://agent-y.com/mcp/",
            protocol="mcp"
        )
    ],
    # Webhook URL template (macros: {agent_id}, {task_type}, {operation_id})
    webhook_url_template="https://myapp.com/webhook/{task_type}/{agent_id}/{operation_id}",

    # Activity callback - fires for ALL events
    on_activity=lambda activity: print(f"[{activity.type}] {activity.task_type}"),

    # Status change handlers
    handlers={
        "on_get_products_status_change": lambda response, metadata: (
            db.save_products(metadata.operation_id, response.products)
            if metadata.status == "completed" else None
        )
    }
) as client:
    # Execute operation - library handles operation IDs, webhook URLs, context management
    agent = client.agent("agent_x")
    request = GetProductsRequest(brief="Coffee brands", buying_mode="brief")
    result = await agent.get_products(request)

    # Check result
    if result.status == "completed":
        # Agent completed synchronously!
        print(f"✅ Sync completion: {len(result.data.products)} products")

    if result.status == "submitted":
        # Agent will send webhook when complete
        print(f"⏳ Async - webhook registered at: {result.submitted.webhook_url}")
# Connections automatically cleaned up here

Documentation

• API Reference - Complete API documentation with type signatures and examples
• Protocol Spec - Ad Context Protocol specification
• Examples - Code examples and usage patterns

The API reference documentation is automatically generated from the code and includes:

• Full type signatures for all methods
• Field descriptions from JSON Schema
• Method documentation with examples
• Searchable interface

Features

Test Helpers

Pre-configured test agents for instant prototyping and testing:

from adcp.testing import (
    test_agent, test_agent_a2a,
    test_agent_no_auth, test_agent_a2a_no_auth,
    creative_agent, test_agent_client, create_test_agent
)
from adcp import GetProductsRequest, PreviewCreativeRequest

# 1. Single agent with authentication (MCP)
result = await test_agent.get_products(
    GetProductsRequest(brief="Coffee brands", buying_mode="brief")
)

# 2. Single agent with authentication (A2A)
result = await test_agent_a2a.get_products(
    GetProductsRequest(brief="Coffee brands", buying_mode="brief")
)

# 3. Single agent WITHOUT authentication (MCP)
# Useful for testing unauthenticated behavior
result = await test_agent_no_auth.get_products(
    GetProductsRequest(brief="Coffee brands", buying_mode="brief")
)

# 4. Single agent WITHOUT authentication (A2A)
result = await test_agent_a2a_no_auth.get_products(
    GetProductsRequest(brief="Coffee brands", buying_mode="brief")
)

# 5. Creative agent (preview functionality, no auth required)
result = await creative_agent.preview_creative(
    PreviewCreativeRequest(
        manifest={"format_id": "banner_300x250", "assets": {...}}
    )
)

# 6. Multi-agent (parallel execution with both protocols)
results = await test_agent_client.get_products(
    GetProductsRequest(brief="Coffee brands", buying_mode="brief")
)

# 7. Custom configuration
from adcp.client import ADCPClient
config = create_test_agent(id="my-test", timeout=60.0)
client = ADCPClient(config)

Use cases:

• Quick prototyping and experimentation
• Example code and documentation
• Integration testing without mock servers
• Testing authentication behavior (comparing auth vs no-auth results)
• Learning AdCP concepts

Important: Test agents are public, rate-limited, and for testing only. Never use in production.

Full Protocol Support

• A2A Protocol: Native support for Agent-to-Agent protocol
• MCP Protocol: Native support for Model Context Protocol
• Auto-detection: Automatically detect which protocol an agent uses

Type Safety

Full type hints with Pydantic validation and auto-generated types from the AdCP spec. All commonly-used types are exported from the main adcp package for convenience:

from adcp import (
    GetProductsRequest,
    BrandManifest,
    Package,
    CpmFixedRatePricingOption,
    MediaBuyStatus,
)

# All methods require typed request objects
request = GetProductsRequest(brief="Coffee brands", buying_mode="brief", max_results=10)
result = await agent.get_products(request)
# result: TaskResult[GetProductsResponse]

if result.success:
    for product in result.data.products:
        print(product.name, product.pricing_options)  # Full IDE autocomplete!

# Type-safe pricing with discriminators
pricing = CpmFixedRatePricingOption(
    pricing_option_id="cpm_usd",
    pricing_model="cpm",
    is_fixed=True,  # Literal[True] - type checked!
    currency="USD",
    rate=5.0
)

# Type-safe status enums
if media_buy.status == MediaBuyStatus.active:
    print("Media buy is active")

Exported from main package:

• Core domain types: BrandManifest, Creative, CreativeManifest, MediaBuy, Package
• Status enums: CreativeStatus, MediaBuyStatus, PackageStatus, PricingModel
• All 9 pricing options: CpcPricingOption, CpmFixedRatePricingOption, VcpmAuctionPricingOption, etc.
• Request/Response types: All 16 operations with full request/response types

Semantic Type Aliases

For discriminated union types (success/error responses), use semantic aliases for clearer code:

from adcp import (
    CreateMediaBuySuccessResponse,  # Clear: this is the success case
    CreateMediaBuyErrorResponse,     # Clear: this is the error case
)

def handle_response(
    response: CreateMediaBuySuccessResponse | CreateMediaBuyErrorResponse
) -> None:
    if isinstance(response, CreateMediaBuySuccessResponse):
        print(f"✅ Media buy created: {response.media_buy_id}")
    else:
        print(f"❌ Errors: {response.errors}")

Available semantic aliases:

• Response types: *SuccessResponse / *ErrorResponse (e.g., CreateMediaBuySuccessResponse)
• Request variants: *FormatRequest / *ManifestRequest (e.g., PreviewCreativeFormatRequest)
• Preview renders: PreviewRenderImage / PreviewRenderHtml / PreviewRenderIframe
• Activation keys: PropertyIdActivationKey / PropertyTagActivationKey

See examples/type_aliases_demo.py for more examples.

Import guidelines:

• ✅ DO: Import from main package: from adcp import GetProductsRequest
• ✅ DO: Use semantic aliases: from adcp import CreateMediaBuySuccessResponse
• ⚠️ AVOID: Import from internal modules: from adcp.types._generated import CreateMediaBuyResponse1

The main package exports provide a stable API while internal generated types may change.

Multi-Agent Operations

Execute across multiple agents simultaneously:

from adcp import GetProductsRequest

# Parallel execution across all agents
request = GetProductsRequest(brief="Coffee brands", buying_mode="brief")
results = await client.get_products(request)

for result in results:
    if result.status == "completed":
        print(f"Sync: {len(result.data.products)} products")
    elif result.status == "submitted":
        print(f"Async: webhook to {result.submitted.webhook_url}")

Webhook Handling

Single endpoint handles all webhooks:

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhook/{task_type}/{agent_id}/{operation_id}")
async def webhook(task_type: str, agent_id: str, operation_id: str, request: Request):
    payload = await request.json()
    payload["task_type"] = task_type
    payload["operation_id"] = operation_id

    # Route to agent client - handlers fire automatically
    agent = client.agent(agent_id)
    await agent.handle_webhook(
        payload,
        request.headers.get("x-adcp-signature")
    )

    return {"received": True}

Security

Webhook signature verification built-in:

client = ADCPMultiAgentClient(
    agents=agents,
    webhook_secret=os.getenv("WEBHOOK_SECRET")
)
# Signatures verified automatically on handle_webhook()

Debug Mode

Enable debug mode to see full request/response details:

agent_config = AgentConfig(
    id="agent_x",
    agent_uri="https://agent-x.com",
    protocol="mcp",
    debug=True  # Enable debug mode
)

result = await client.agent("agent_x").get_products(brief="Coffee brands", buying_mode="brief")

# Access debug information
if result.debug_info:
    print(f"Duration: {result.debug_info.duration_ms}ms")
    print(f"Request: {result.debug_info.request}")
    print(f"Response: {result.debug_info.response}")

Or use the CLI:

uvx adcp --debug myagent get_products '{"brief":"TV ads"}'

Resource Management

Why use async context managers?

• Ensures HTTP connections are properly closed, preventing resource leaks
• Handles cleanup even when exceptions occur
• Required for production applications with connection pooling
• Prevents issues with async task group cleanup in MCP protocol

The recommended pattern uses async context managers:

from adcp import ADCPClient, AgentConfig, GetProductsRequest

# Recommended: Automatic cleanup with context manager
config = AgentConfig(id="agent_x", agent_uri="https://...", protocol="a2a")
async with ADCPClient(config) as client:
    request = GetProductsRequest(brief="Coffee brands", buying_mode="brief")
    result = await client.get_products(request)
    # Connection automatically closed on exit

# Multi-agent client also supports context managers
async with ADCPMultiAgentClient(agents) as client:
    # Execute across all agents in parallel
    results = await client.get_products(request)
    # All agent connections closed automatically (even if some failed)

Manual cleanup is available for special cases (e.g., managing client lifecycle manually):

# Use manual cleanup when you need fine-grained control over lifecycle
client = ADCPClient(config)
try:
    result = await client.get_products(request)
finally:
    await client.close()  # Explicit cleanup

When to use manual cleanup:

• Managing client lifecycle across multiple functions
• Testing scenarios requiring explicit control
• Integration with frameworks that manage resources differently

In most cases, prefer the context manager pattern.

Error Handling

The library provides a comprehensive exception hierarchy with helpful error messages:

from adcp.exceptions import (
    ADCPError,               # Base exception
    ADCPConnectionError,     # Connection failed
    ADCPAuthenticationError, # Auth failed (401, 403)
    ADCPTimeoutError,        # Request timed out
    ADCPProtocolError,       # Invalid response format
    ADCPToolNotFoundError,   # Tool not found
    ADCPWebhookSignatureError  # Invalid webhook signature
)

try:
    result = await client.agent("agent_x").get_products(brief="Coffee", buying_mode="brief")
except ADCPAuthenticationError as e:
    # Exception includes agent context and helpful suggestions
    print(f"Auth failed for {e.agent_id}: {e.message}")
    print(f"Suggestion: {e.suggestion}")
except ADCPTimeoutError as e:
    print(f"Request timed out after {e.timeout}s")
except ADCPConnectionError as e:
    print(f"Connection failed: {e.message}")
    print(f"Agent URI: {e.agent_uri}")
except ADCPError as e:
    # Catch-all for other AdCP errors
    print(f"AdCP error: {e.message}")

All exceptions include:

• Contextual information: agent ID, URI, and operation details
• Actionable suggestions: specific steps to fix common issues
• Error classification: proper HTTP status code handling

Available Tools

All AdCP tools with full type safety:

Media Buy Lifecycle:

• get_products() - Discover advertising products
• list_creative_formats() - Get supported creative formats
• create_media_buy() - Create new media buy
• update_media_buy() - Update existing media buy
• sync_creatives() - Upload/sync creative assets
• list_creatives() - List creative assets
• get_media_buy_delivery() - Get delivery performance

Creative Management:

• preview_creative() - Preview creative before building
• build_creative() - Generate production-ready creative assets

Discovery & Accounts:

• get_adcp_capabilities() - Discover agent capabilities and authorized publishers
• list_accounts() - List billing accounts

Audience & Targeting:

• get_signals() - Get audience signals
• activate_signal() - Activate audience signals
• provide_performance_feedback() - Send performance feedback

Workflow Examples

Complete Media Buy Workflow

A typical media buy workflow involves discovering products, creating the buy, and managing creatives:

from adcp import ADCPClient, AgentConfig, GetProductsRequest, CreateMediaBuyRequest
from adcp import BrandManifest, PublisherPropertiesAll

# 1. Connect to agent
config = AgentConfig(id="sales_agent", agent_uri="https://...", protocol="mcp")
async with ADCPClient(config) as client:

    # 2. Discover available products
    products_result = await client.get_products(
        GetProductsRequest(brief="Premium video inventory for coffee brand", buying_mode="brief")
    )

    if products_result.success:
        product = products_result.data.products[0]
        print(f"Found product: {product.name}")

    # 3. Create media buy reservation
    media_buy_result = await client.create_media_buy(
        CreateMediaBuyRequest(
            brand_manifest=BrandManifest(
                name="Coffee Co",
                brand_url="https://coffeeco.com",
                logo_url="https://coffeeco.com/logo.png",
                # ... additional brand details
            ),
            packages=[{
                "package_id": product.packages[0].package_id,
                "quantity": 1000000  # impressions
            }],
            publisher_properties=PublisherPropertiesAll(
                selection_type="all"  # Target all authorized properties
            )
        )
    )

    if media_buy_result.success:
        media_buy_id = media_buy_result.data.media_buy_id
        print(f"✅ Media buy created: {media_buy_id}")

    # 4. Update media buy if needed
    from adcp import UpdateMediaBuyPackagesRequest

    update_result = await client.update_media_buy(
        UpdateMediaBuyPackagesRequest(
            media_buy_id=media_buy_id,
            packages=[{
                "package_id": product.packages[0].package_id,
                "quantity": 1500000  # Increase budget
            }]
        )
    )

    if update_result.success:
        print("✅ Media buy updated")

Complete Creative Workflow

Build and deliver production-ready creatives:

from adcp import ADCPClient, AgentConfig
from adcp import PreviewCreativeFormatRequest, BuildCreativeRequest
from adcp import CreativeManifest, PlatformDeployment

# 1. Connect to creative agent
config = AgentConfig(id="creative_agent", agent_uri="https://...", protocol="mcp")
async with ADCPClient(config) as client:

    # 2. List available formats
    formats_result = await client.list_creative_formats()

    if formats_result.success:
        format_id = formats_result.data.formats[0].format_id
        print(f"Using format: {format_id.id}")

    # 3. Preview creative (test before building)
    preview_result = await client.preview_creative(
        PreviewCreativeFormatRequest(
            target_format_id=format_id.id,
            inputs={
                "headline": "Fresh Coffee Daily",
                "cta": "Order Now"
            },
            output_format="url"  # Get preview URL
        )
    )

    if preview_result.success:
        preview_url = preview_result.data.renders[0].url
        print(f"Preview at: {preview_url}")

    # 4. Build production creative
    build_result = await client.build_creative(
        BuildCreativeRequest(
            manifest=CreativeManifest(
                format_id=format_id,
                brand_url="https://coffeeco.com",
                # ... creative content
            ),
            target_format_id=format_id.id,
            deployment=PlatformDeployment(
                type="platform",
                platform_id="google_admanager"
            )
        )
    )

    if build_result.success:
        vast_url = build_result.data.assets[0].url
        print(f"✅ Creative ready: {vast_url}")

Integrated Workflow: Media Buy + Creatives

Combine both workflows for a complete campaign setup:

from adcp import ADCPMultiAgentClient, AgentConfig
from adcp import GetProductsRequest, CreateMediaBuyRequest, BuildCreativeRequest

# Connect to both sales and creative agents
async with ADCPMultiAgentClient(
    agents=[
        AgentConfig(id="sales", agent_uri="https://sales-agent.com", protocol="mcp"),
        AgentConfig(id="creative", agent_uri="https://creative-agent.com", protocol="mcp"),
    ]
) as client:

    # 1. Get products from sales agent
    sales_agent = client.agent("sales")
    products = await sales_agent.simple.get_products(
        brief="Premium video inventory",
        buying_mode="brief",
    )

    # 2. Get creative formats from creative agent
    creative_agent = client.agent("creative")
    formats = await creative_agent.simple.list_creative_formats()

    # 3. Build creative asset
    creative_result = await creative_agent.build_creative(
        BuildCreativeRequest(
            manifest=creative_manifest,
            target_format_id=formats.formats[0].format_id.id
        )
    )

    # 4. Create media buy with creative
    media_buy_result = await sales_agent.create_media_buy(
        CreateMediaBuyRequest(
            brand_manifest=brand_manifest,
            packages=[{"package_id": products.products[0].packages[0].package_id}],
            publisher_properties=publisher_properties,
            creative_urls=[creative_result.data.assets[0].url]
        )
    )

    print(f"✅ Campaign live: {media_buy_result.data.media_buy_id}")

Property Discovery (AdCP v2.2.0)

Build agent registries by discovering properties agents can sell:

from adcp.discovery import PropertyCrawler, get_property_index

# Crawl agents to discover properties
crawler = PropertyCrawler()
await crawler.crawl_agents([
    {"agent_url": "https://agent-x.com", "protocol": "a2a"},
    {"agent_url": "https://agent-y.com/mcp/", "protocol": "mcp"}
])

index = get_property_index()

# Query 1: Who can sell this property?
matches = index.find_agents_for_property("domain", "cnn.com")

# Query 2: What can this agent sell?
auth = index.get_agent_authorizations("https://agent-x.com")

# Query 3: Find by tags
premium = index.find_agents_by_property_tags(["premium", "ctv"])

Publisher Authorization Validation

Verify sales agents are authorized to sell publisher properties via adagents.json:

from adcp import (
    fetch_adagents,
    verify_agent_authorization,
    verify_agent_for_property,
)

# Fetch and parse adagents.json from publisher
adagents_data = await fetch_adagents("publisher.com")

# Verify agent authorization for a property
is_authorized = verify_agent_authorization(
    adagents_data=adagents_data,
    agent_url="https://sales-agent.example.com",
    property_type="website",
    property_identifiers=[{"type": "domain", "value": "publisher.com"}]
)

# Or use convenience wrapper (fetch + verify in one call)
is_authorized = await verify_agent_for_property(
    publisher_domain="publisher.com",
    agent_url="https://sales-agent.example.com",
    property_identifiers=[{"type": "domain", "value": "publisher.com"}],
    property_type="website"
)

Domain Matching Rules:

• Exact match: example.com matches example.com
• Common subdomains: www.example.com matches example.com
• Wildcards: api.example.com matches *.example.com
• Protocol-agnostic: http://agent.com matches https://agent.com

Use Cases:

• Sales agents verify authorization before accepting media buys
• Publishers test their adagents.json files
• Developer tools build authorization validators

See examples/adagents_validation.py for complete examples.

Authorization Discovery

Discover which publishers have authorized your agent using two approaches:

1. "Push" Approach - Ask the agent (recommended, fastest):

from adcp import ADCPClient, GetAdcpCapabilitiesRequest

async with ADCPClient(agent_config) as client:
    # Single API call to agent
    result = await client.get_adcp_capabilities(GetAdcpCapabilitiesRequest())
    if result.success and result.data.media_buy:
        portfolio = result.data.media_buy.portfolio
        print(f"Authorized for: {portfolio.publisher_domains}")

2. "Pull" Approach - Check publisher adagents.json files (when you need property details):

from adcp import fetch_agent_authorizations

# Check specific publishers (fetches in parallel)
contexts = await fetch_agent_authorizations(
    "https://our-sales-agent.com",
    ["nytimes.com", "wsj.com", "cnn.com"]
)

for domain, ctx in contexts.items():
    print(f"{domain}:")
    print(f"  Property IDs: {ctx.property_ids}")
    print(f"  Tags: {ctx.property_tags}")

When to use which:

• Push: Quick discovery, portfolio overview, high-level authorization check
• Pull: Property-level details, specific publisher list, works offline

See examples/fetch_agent_authorizations.py for complete examples.

CLI Tool

The adcp command-line tool provides easy interaction with AdCP agents without writing code.

Installation

# Install globally
pip install adcp

# Or use uvx to run without installing
uvx adcp --help

Quick Start

# Save agent configuration
uvx adcp --save-auth myagent https://agent.example.com mcp

# List tools available on agent
uvx adcp myagent list_tools

# Execute a tool
uvx adcp myagent get_products '{"brief":"TV ads"}'

# Use from stdin
echo '{"brief":"TV ads"}' | uvx adcp myagent get_products

# Use from file
uvx adcp myagent get_products @request.json

# Get JSON output
uvx adcp --json myagent get_products '{"brief":"TV ads"}'

# Enable debug mode
uvx adcp --debug myagent get_products '{"brief":"TV ads"}'

Using Test Agents from CLI

The CLI provides easy access to public test agents without configuration:

# Use test agent with authentication (MCP)
uvx adcp https://test-agent.adcontextprotocol.org/mcp/ \
  --auth 1v8tAhASaUYYp4odoQ1PnMpdqNaMiTrCRqYo9OJp6IQ \
  get_products '{"brief":"Coffee brands"}'

# Use test agent WITHOUT authentication (MCP)
uvx adcp https://test-agent.adcontextprotocol.org/mcp/ \
  get_products '{"brief":"Coffee brands"}'

# Use test agent with authentication (A2A)
uvx adcp --protocol a2a \
  --auth 1v8tAhASaUYYp4odoQ1PnMpdqNaMiTrCRqYo9OJp6IQ \
  https://test-agent.adcontextprotocol.org \
  get_products '{"brief":"Coffee brands"}'

# Save test agent for easier access
uvx adcp --save-auth test-agent https://test-agent.adcontextprotocol.org/mcp/ mcp
# Enter token when prompted: 1v8tAhASaUYYp4odoQ1PnMpdqNaMiTrCRqYo9OJp6IQ

# Now use saved config
uvx adcp test-agent get_products '{"brief":"Coffee brands"}'

# Use creative agent (no auth required)
uvx adcp https://creative.adcontextprotocol.org/mcp \
  preview_creative @creative_manifest.json

Test Agent Details:

• URL (MCP): https://test-agent.adcontextprotocol.org/mcp/
• URL (A2A): https://test-agent.adcontextprotocol.org
• Auth Token: 1v8tAhASaUYYp4odoQ1PnMpdqNaMiTrCRqYo9OJp6IQ (optional, public token)
• Rate Limited: For testing only, not for production
• No Auth Mode: Omit --auth flag to test unauthenticated behavior

### Configuration Management

```bash
# Save agent with authentication
uvx adcp --save-auth myagent https://agent.example.com mcp
# Prompts for optional auth token

# List saved agents
uvx adcp --list-agents

# Remove saved agent
uvx adcp --remove-agent myagent

# Show config file location
uvx adcp --show-config

Direct URL Access

# Use URL directly without saving
uvx adcp https://agent.example.com/mcp list_tools

# Override protocol
uvx adcp --protocol a2a https://agent.example.com list_tools

# Pass auth token
uvx adcp --auth YOUR_TOKEN https://agent.example.com list_tools

Examples

# Get products from saved agent
uvx adcp myagent get_products '{"brief":"Coffee brands for digital video"}'

# Create media buy
uvx adcp myagent create_media_buy '{
  "name": "Q4 Campaign",
  "budget": 50000,
  "start_date": "2024-01-01",
  "end_date": "2024-03-31"
}'

# List creative formats with JSON output
uvx adcp --json myagent list_creative_formats | jq '.data'

# Debug connection issues
uvx adcp --debug myagent list_tools

Configuration File

Agent configurations are stored in ~/.adcp/config.json:

{
  "agents": {
    "myagent": {
      "agent_uri": "https://agent.example.com",
      "protocol": "mcp",
      "auth_token": "optional-token"
    }
  }
}

Environment Configuration

# .env
WEBHOOK_URL_TEMPLATE="https://myapp.com/webhook/{task_type}/{agent_id}/{operation_id}"
WEBHOOK_SECRET="your-webhook-secret"

ADCP_AGENTS='[
  {
    "id": "agent_x",
    "agent_uri": "https://agent-x.com",
    "protocol": "a2a",
    "auth_token_env": "AGENT_X_TOKEN"
  }
]'
AGENT_X_TOKEN="actual-token-here"

# Auto-discover from environment
client = ADCPMultiAgentClient.from_env()

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy src/

# Format code
black src/ tests/
ruff check src/ tests/

Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

License

Apache 2.0 License - see LICENSE file for details.

Support

• API Reference: adcontextprotocol.github.io/adcp-client-python
• Protocol Documentation: docs.adcontextprotocol.org
• Issues: GitHub Issues
• Protocol Spec: AdCP Specification