OpenAdTrace Python SDK

Python client for the OpenAdTrace MCP server. Zero external dependencies.

---

Install

pip install openadtrace-sdk

---

Setup

You need a running OpenAdTrace server. Quickest way:

# Clone and start with Docker
git clone https://github.com/Samrajtheailyceum/open-ad-trace.git
cd open-ad-trace && docker compose up -d

# Verify it's running
curl http://localhost:8787/health

Or without Docker:

cd server && npm install && npm run migrate:local && npx wrangler dev

---

Quick Start

from openadtrace_sdk import OpenAdTraceClient

with OpenAdTraceClient("http://localhost:8787/mcp") as client:
    # Record an event
    result = client.trace_event(
        protocol="openrtb",
        layer="bid_request",
        actor_role="dsp",
        actor_id="my-dsp-001",
        event_type="bid",
        action="submit_bid",
        status="success",
        latency_ms=12,
        bid_price_cpm=4.50,
    )
    print(f"Recorded: trace_id={result.trace_id}, span_id={result.span_id}")

The client handles MCP session management automatically — no setup, no handshake code, no reconnection logic.

---

Async Quick Start

import asyncio
from openadtrace_sdk import AsyncOpenAdTraceClient

async def main():
    async with AsyncOpenAdTraceClient("http://localhost:8787/mcp") as client:
        result = await client.trace_event(
            protocol="openrtb",
            layer="bid_request",
            actor_role="dsp",
            actor_id="my-dsp-001",
            event_type="bid",
            action="submit_bid",
            status="success",
        )
        print(result.trace_id)

asyncio.run(main())

---

Authentication

If your server has OAT_API_TOKEN set:

client = OpenAdTraceClient("http://localhost:8787/mcp", token="your-api-token")

---

Real-World Integration Examples

DSP — Log every bid

import time
from openadtrace_sdk import OpenAdTraceClient

with OpenAdTraceClient("http://localhost:8787/mcp", token="your-token") as oat:

    # Your existing bid logic
    start = time.time()
    bid_result = your_bidder.submit_bid(bid_request)
    latency = int((time.time() - start) * 1000)

    # Add one call to create an audit trail
    oat.trace_event(
        protocol="openrtb",
        layer="bid_request",
        actor_role="dsp",
        actor_id="your-dsp-id",
        event_type="bid",
        action="submit_bid",
        status="success" if bid_result.won else "failure",
        latency_ms=latency,
        bid_price_cpm=bid_result.price,
        campaign_id=bid_result.campaign_id,
        rationale_visible="Floor price met, brand safety passed",
    )

SSP — Log supply curation decisions

oat.trace_event(
    protocol="openrtb",
    layer="auction",
    actor_role="ssp",
    actor_id="your-ssp-id",
    event_type="decision",
    action="curate_supply",
    status="success",
    rationale_visible="Filtered 12 bid requests to 3 premium-only opportunities",
)

Agency — Log campaign activation

oat.trace_event(
    protocol="adcp",
    layer="execution",
    actor_role="buyer_agent",
    actor_id="agency-agent-001",
    event_type="decision",
    action="activate_campaign",
    status="success",
    campaign_id="client-campaign-q4",
    media_cost=25000.00,
    currency="USD",
    rationale_visible="Activated CTV campaign per client-approved media plan",
    policy_basis="governance_check: PASSED",
)

---

All 11 Methods

Method	What it does
trace_event(...)	Record an advertising event
validate_event(...)	Check an event against schema without storing
query_traces(...)	Search events by actor, campaign, time range
analyse_traces(...)	Get quality scores and insights
redact_event(...)	Strip PII and secrets from an event
scan_protocols(...)	Check domains for protocol discovery files
verify_supply_chain(...)	Validate ads.txt / sellers.json
trust_score(...)	Get reliability score (0-100) for an actor
correlate_traces(...)	Find related events across agents
compliance_check(...)	Check against protocol guidelines
risk_signals(...)	Detect risk flags for a campaign or actor

---

Builder Pattern

For complex events, use the fluent builder:

from openadtrace_sdk import TraceEventBuilder, Protocol, Layer, Status

event = (
    TraceEventBuilder()
    .protocol(Protocol.OPENRTB)
    .layer(Layer.BID_DECISION)
    .actor("dsp", "my-dsp-001")
    .event("bid", "evaluate")
    .status(Status.SUCCESS)
    .with_campaign("camp-q4-001")
    .latency(8)
    .bid_price(4.50)
    .bid_floor(2.00)
    .rationale("CPM within target range and brand-safe context")
    .build()
)

with OpenAdTraceClient("http://localhost:8787/mcp") as client:
    result = client.trace_event(**event)

---

Batch Requests

Send up to 20 calls in a single HTTP request:

with OpenAdTraceClient("http://localhost:8787/mcp") as client:
    results = (
        client.batch()
        .trace_event(
            protocol="openrtb", layer="bid_request",
            actor_role="dsp", actor_id="my-dsp-001",
            event_type="bid", action="submit_bid", status="success",
        )
        .trust_score(actor_id="my-dsp-001")
        .risk_signals(actor_id="my-dsp-001", period_days=7)
        .execute()
    )

    trace_result, trust_result, risk_result = results
    print(f"Trust score: {trust_result.trust_score}")

---

Error Handling

from openadtrace_sdk import (
    OpenAdTraceClient, OpenAdTraceError,
    AuthenticationError, RateLimitError, NetworkError, ToolError,
)

try:
    with OpenAdTraceClient("http://localhost:8787/mcp", token="key") as client:
        result = client.trust_score(actor_id="my-dsp-001")
except AuthenticationError:
    print("Invalid API token")
except RateLimitError as e:
    print(f"Rate limited — retry after {e.retry_after_seconds}s")
except NetworkError as e:
    print(f"Connection problem: {e}")
except ToolError as e:
    print(f"Tool '{e.tool_name}' failed: {e}")
except OpenAdTraceError as e:
    print(f"SDK error: {e}")

---

Enums (optional)

Use enums for type safety, or just pass strings — both work:

from openadtrace_sdk import Protocol, Layer, Status

# These two calls are identical:
client.trace_event(protocol=Protocol.OPENRTB, layer=Layer.BID_REQUEST, status=Status.SUCCESS, ...)
client.trace_event(protocol="openrtb", layer="bid_request", status="success", ...)

---

Enterprise Features

Context Signals — brand safety, audience, viewability, attention

from openadtrace_sdk.context import ContextSignals, GARMFloor, GARMCategory

ctx = (
    ContextSignals()
    .brand_safety(floor=GARMFloor.HIGH, exclude=[GARMCategory.ALCOHOL, GARMCategory.GAMBLING],
                  provider="ias", verdict="brand_safe", score=92.0)
    .audience(segments=["auto_intenders", "luxury_travel"], min_age=25, first_party=True)
    .viewability(target_rate=0.70, provider="moat")
    .attention(target_seconds=8.0, provider="adelaide")
    .geo(country="US", region="northeast", dma="501")
    .device(types=["ctv", "mobile"])
    .content(category="entertainment", publisher="espn.com")
    .supply_path(hops=1, direct=True)
)

# Attach to any trace event
client.trace_event(..., meta=ctx.to_dict())

Campaign Lifecycle Tracking

from openadtrace_sdk.campaign import CampaignTracker

campaign = CampaignTracker(
    client,
    campaign_id="brand-q4-2025",
    advertiser="Acme Corp",
    brand="Acme Premium",
    budget=500_000.00,
    currency="USD",
    context=ctx,  # auto-attached to every event
)

campaign.plan("CTV + premium digital, 25-54 auto-intenders")
campaign.activate()
campaign.log_spend(25_000, vendor="the-trade-desk", channel="ctv")
campaign.log_spend(15_000, vendor="dv360", channel="display")
campaign.bid(price_cpm=4.50, floor_cpm=2.00, won=True)
campaign.governance_check("brand_safety", passed=True, details="All placements verified")
campaign.complete()

report = campaign.summary()
# {'total_spend': 40000, 'budget_remaining': 460000, 'spend_by_vendor': {...}}

Governance Workflows — human approval chains

from openadtrace_sdk.governance import GovernanceWorkflow, ApprovalGate

gov = GovernanceWorkflow(client, workflow_id="launch-approval", campaign_id="brand-q4")
gov.add_gate(ApprovalGate(name="brand_safety", approver="brand_team", required=True))
gov.add_gate(ApprovalGate(name="age_gate", approver="compliance", regulations=["ftc", "ttb"]))
gov.add_gate(ApprovalGate(name="budget", approver="finance", threshold=100_000))

gov.approve("brand_safety", approved_by="jane@brand.com", notes="All clear")
gov.approve("age_gate", approved_by="compliance@brand.com", notes="21+ confirmed")
gov.approve("budget", approved_by="cfo@brand.com")

if gov.all_approved():
    campaign.activate()

Multi-Market Support — 22 markets with regulatory presets

from openadtrace_sdk.markets import get_market, markets_restricting, list_markets

us = get_market("US")   # alcohol_min_age=21, regulations=['ftc','coppa','ccpa',...]
uk = get_market("GB")   # alcohol_min_age=18, regulations=['uk_gdpr','asa','ofcom',...]
jp = get_market("JP")   # alcohol_min_age=20, regulations=['appi','jaro']

# All markets restricting alcohol
markets_restricting("alcohol")   # 18 markets

# Verify age-gating
us.age_gate_check(min_age=21)    # True
us.age_gate_check(min_age=18)    # False — US requires 21+

Available markets: US, CA, MX, GB, DE, FR, ES, IT, NL, SE, JP, AU, SG, IN, KR, CN, AE, SA, ZA, NG, BR, AR.

---

Requirements

• Python 3.9+
• No external dependencies (stdlib only)

License

Apache-2.0. See the main repository for details.