Run a PydanticAI agent

Purpose

Run your own Python process with governed model calls, MCP tools, and telemetry using the repository's mission_control_pydantic_ai adapter. This adapter targets PydanticAI 2.31.0; acquire it from a Mission Control source checkout. This guide does not assume a published package index release.

Prerequisites

• Python 3.12–3.14 and uv, and a checkout of the Mission Control release you operate.
• An ordinary user or service-account virtual key, held only by your process, with model.generate and telemetry.ingest.
• A generation model allowed by catalog, Team allowance and key policy. Optional MCP tools must also be enabled and allowed.
• The inference origin and Collector HTTP endpoint from your operator. These can be different origins; the Collector endpoint ends in /v1/traces.

Procedure

1. From the repository root, install the locked adapter environment:

cd integrations/pydantic-ai
uv sync --locked --all-groups

For an existing consumer environment, install the source directory with python -m pip install /path/to/mission-control/integrations/pydantic-ai. The locked environment above is the reproducible reference setup.

1. In that same working directory, set your deployment values. Replace the fictional key with a key issued in the Console:

export MISSION_CONTROL_GATEWAY_URL=https://gateway.example.test
export MISSION_CONTROL_API_KEY=mck_example_not_a_real_key
export MISSION_CONTROL_VIRTUAL_MODEL=support
export MISSION_CONTROL_OTLP_ENDPOINT=https://collector.example.test/v1/traces

1. Save the following as consumer.py in integrations/pydantic-ai. It makes one model request; it needs no fixture tool or YAML definition:

import asyncio
import json
import os
import secrets

import httpx
from pydantic import SecretStr
from pydantic_ai import Agent
from mission_control_pydantic_ai import (
    ExternalObservationContext, MissionControlConfig,
    MissionControlOtlpSpanExporter, attach_execution,
    create_pydantic_model, install_pydantic_ai_instrumentation,
)

async def main():
    config = MissionControlConfig(
        gateway_url=os.environ["MISSION_CONTROL_GATEWAY_URL"],
        api_key=SecretStr(os.environ["MISSION_CONTROL_API_KEY"]),
        virtual_model=os.environ["MISSION_CONTROL_VIRTUAL_MODEL"],
        otlp_endpoint=os.environ["MISSION_CONTROL_OTLP_ENDPOINT"],
        otlp_protocol="http/protobuf",
    )
    trace_id, parent_id = secrets.token_hex(16), secrets.token_hex(8)
    execution = ExternalObservationContext(
        observation_id=secrets.token_urlsafe(24),
        trace_id=trace_id, parent_span_id=parent_id,
        traceparent=f"00-{trace_id}-{parent_id}-01",
        telemetry_capture_mode="metadata",
        observation_schema_version="mission-control.metadata/v1",
    )
    exporter = MissionControlOtlpSpanExporter(config)
    provider = install_pydantic_ai_instrumentation(exporter)
    try:
        async with attach_execution(execution):
            try:
                async with httpx.AsyncClient(timeout=30) as client:
                    model = create_pydantic_model(
                        config, execution, http_client=client,
                    )
                    result = await Agent(model).run("Say hello in one sentence.", model_settings={"max_tokens": 128})
                    print(result.output)
            finally:
                complete = await exporter.flush(execution)
                print(json.dumps({"traceId": trace_id,
                                  "telemetryStatus": "complete" if complete else "partial"}))
    finally:
        provider.shutdown()

asyncio.run(main())

1. Run uv run --locked python consumer.py. A successful run prints the model's greeting and JSON with a 32-character Trace ID and an independent telemetryStatus.
2. Open Observe → Traces, select the Team and time range, and open that Trace ID. Check the model request's decision and the PydanticAI spans.

How the pieces fit

Create one MissionControlConfig and instrumentation provider per process, and a new ExternalObservationContext for each application execution. Its observation ID is local adapter bookkeeping; it does not register a public lifecycle resource. The actual virtual key determines authority. An observation ID or caller span attributes cannot select a Team, namespace, payer or verified person.

Both create_pydantic_model(config, execution) and create_mcp_toolset(config, execution) require the execution argument. attach_execution(execution) propagates W3C context asynchronously. To expose governed tools to the model, create a toolset and pass it as Agent(model, toolsets=[toolset]); use async with agent: around agent.run(...) and close the toolset with await toolset.aclose() in a finally block. Tool schemas and permissions come from the gateway; see the MCP contract.

Run the example

The source checkout also contains the support agent program at integrations/pydantic-ai/examples/support_agent.py, a deterministic conformance program using examples/agent.yaml. From integrations/pydantic-ai, keep the environment above, set MISSION_CONTROL_MCP_TOOL to a permitted tool accepting { "query": "..." }, set MISSION_CONTROL_OTLP_PROTOCOL=http/protobuf, and run uv run --locked python examples/support_agent.py. It runs a model request, directly calls the tool, and emits synthetic content spans with local full capture. Success prints status: "completed", traceId, observationId, and telemetryStatus; it is a fixture journey, not the minimal consumer above.

Telemetry behavior

Only completed spans are exported. Local metadata strips content; local full retains permitted content with hard exclusions. Collector policy remains authoritative and can remove more. The exporter batches at most 64 spans and buffers at most 20,000 per execution. Overflow, sanitization, and export failures report incomplete telemetry independently of the model result. The default detected service name is mission-control-pydantic-ai.

Limits

Reusable SDK clients impose no global output bound. The example above explicitly bounds each call; the support conformance example uses 4,096 output tokens per round. Under finite caps, unbounded/unknown liability fails closed.

HTTP/protobuf endpoints require HTTPS and /v1/traces; gRPC uses an HTTPS authority without a path. Loopback HTTP is allowed for development. Configuration changes require a new process and execution context. No Agent registration/publication is required for direct model/MCP/OTLP work. Managed deployment invocation uses its separate admission contract.

Verification

From integrations/pydantic-ai, run uv run --locked pytest -q for adapter protocol, context, sanitization, and export checks. Confirm the live request decision and matching Trace in your deployment separately.

Failure recovery

For denied calls, check key revocation, key/owner/Team state, Team allowance, key policy, and budgets using Gateway errors. For partial telemetry, check the Collector endpoint, protocol, credential, and availability. A successful agent result does not prove export delivery. Never retry model work solely to repair telemetry. Changing local capture cannot restore content already discarded; see capture policy.

Related links

• Run external code
• Instrument with OpenTelemetry
• API reference