Skip to main content

​
Overview

Use respan-exporter-pydantic-ai to instrument Pydantic AI agents with Respan. Traces, spans, and metrics from agent runs (including LLM calls and tool use) are sent to Respan via OpenTelemetry and standard semantic conventions. The exporter requires respan-tracing (installed automatically). You can route LLM calls through Respan as the gateway, so only your Respan API key is needed—no separate OpenAI key.

​
Setup

1. Install 
pip install pydantic-ai respan-exporter-pydantic-ai
2. Environment Set RESPAN_API_KEY (required). Optional: RESPAN_BASE_URL (default https://api.respan.ai/api). For the gateway pattern, also set OPENAI_BASE_URL and OPENAI_API_KEY to your Respan base URL and API key so no separate OpenAI key is needed.

​
Examples

Each example below is a complete runnable script. Run it and open the Respan dashboard to see the trace.

​
Example: Hello World

Minimal instrumentation: initialize Respan, instrument Pydantic AI, run one agent call. 
import os

# Route LLM calls through Respan (gateway pattern — only RESPAN_API_KEY needed)
respan_api_key = os.environ["RESPAN_API_KEY"]
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
os.environ["OPENAI_BASE_URL"] = respan_base_url
os.environ["OPENAI_API_KEY"] = respan_api_key

from pydantic_ai import Agent
from respan_tracing import RespanTelemetry
from respan_exporter_pydantic_ai import instrument_pydantic_ai

def main():
    telemetry = RespanTelemetry(
        app_name="pydantic-ai-hello-world",
        api_key=respan_api_key,
        base_url=respan_base_url,
    )
    instrument_pydantic_ai()

    agent = Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant.",
    )
    result = agent.run_sync("What is the capital of France?")
    print("Agent Output:", result.output)

    telemetry.flush()

if __name__ == "__main__":
    main()
hello_world

​
Example: Gateway

Use Respan as the LLM gateway and control content capture with include_content and include_binary_content. 
import os

respan_api_key = os.environ["RESPAN_API_KEY"]
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
os.environ["OPENAI_BASE_URL"] = respan_base_url
os.environ["OPENAI_API_KEY"] = respan_api_key

from pydantic_ai import Agent
from respan_tracing import RespanTelemetry
from respan_exporter_pydantic_ai import instrument_pydantic_ai

def main():
    telemetry = RespanTelemetry(
        app_name="pydantic-ai-gateway",
        api_key=respan_api_key,
        base_url=respan_base_url,
    )
    instrument_pydantic_ai(
        include_content=True,
        include_binary_content=True,
    )

    agent = Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant.",
    )
    result = agent.run_sync("What is the capital of France?")
    print("Agent Output:", result.output)

    telemetry.flush()

if __name__ == "__main__":
    main()
gateway

​
Example: Workflows and tasks

Use @workflow and @task from respan-tracing to group spans into workflows and tasks. 
import os

respan_api_key = os.environ["RESPAN_API_KEY"]
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
os.environ["OPENAI_BASE_URL"] = respan_base_url
os.environ["OPENAI_API_KEY"] = respan_api_key

from pydantic_ai import Agent
from respan_tracing import RespanTelemetry
from respan_tracing.decorators import workflow, task
from respan_exporter_pydantic_ai import instrument_pydantic_ai

agent = Agent("openai:gpt-4o", system_prompt="You are a helpful travel assistant.")

@task(name="fetch_destination_info")
def fetch_destination_info(destination: str) -> str:
    result = agent.run_sync(f"Give me a one-sentence summary of {destination}.")
    return result.output

@workflow(name="travel_planning_workflow")
def travel_planning_workflow(destination: str):
    info = fetch_destination_info(destination)
    return info

def main():
    telemetry = RespanTelemetry(
        app_name="pydantic-ai-tracing",
        api_key=respan_api_key,
        base_url=respan_base_url,
    )
    instrument_pydantic_ai()
    output = travel_planning_workflow("Paris")
    print("Workflow Output:", output)
    telemetry.flush()

if __name__ == "__main__":
    main()
tracing

​
Example: Respan params on spans

Set customer_identifier, metadata, and custom_tags on the current span with get_client().update_current_span(respan_params=...). 
import os

respan_api_key = os.environ["RESPAN_API_KEY"]
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
os.environ["OPENAI_BASE_URL"] = respan_base_url
os.environ["OPENAI_API_KEY"] = respan_api_key

from pydantic_ai import Agent
from respan_tracing import RespanTelemetry, get_client
from respan_tracing.decorators import task
from respan_exporter_pydantic_ai import instrument_pydantic_ai

agent = Agent("openai:gpt-4o")

@task(name="customer_query")
def customer_query(query: str):
    client = get_client()
    if client:
        client.update_current_span(
            respan_params={
                "customer_identifier": "user_12345",
                "metadata": {"plan": "premium", "session_id": "abc-987"},
                "custom_tags": ["pydantic-ai", "test-run"],
            }
        )
    result = agent.run_sync(query)
    return result.output

def main():
    telemetry = RespanTelemetry(
        app_name="pydantic-ai-params",
        api_key=respan_api_key,
        base_url=respan_base_url,
    )
    instrument_pydantic_ai()
    output = customer_query("Hello, who are you?")
    print("Output:", output)
    telemetry.flush()

if __name__ == "__main__":
    main()
respan_params

​
Example: Tool use

Agents that use tools are traced automatically; tool calls and results appear as spans. 
import os

respan_api_key = os.environ["RESPAN_API_KEY"]
respan_base_url = os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api")
os.environ["OPENAI_BASE_URL"] = respan_base_url
os.environ["OPENAI_API_KEY"] = respan_api_key

from pydantic_ai import Agent
from respan_tracing import RespanTelemetry
from respan_tracing.decorators import workflow
from respan_exporter_pydantic_ai import instrument_pydantic_ai

agent = Agent(
    "openai:gpt-4o",
    system_prompt=(
        "You are a calculator assistant. You must use the provided tools for any arithmetic. "
        "Never compute numbers yourself; always call the add tool when asked to add numbers."
    ),
)

@agent.tool_plain
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@workflow(name="calculator_agent_run")
def run_calculator_agent(prompt: str):
    result = agent.run_sync(prompt)
    return result.output

def main():
    telemetry = RespanTelemetry(
        app_name="pydantic-ai-tool-use",
        api_key=respan_api_key,
        base_url=respan_base_url,
    )
    instrument_pydantic_ai()
    output = run_calculator_agent("Use your add tool to compute 15 + 27, then reply with the result.")
    print("Agent Output:", output)
    telemetry.flush()

if __name__ == "__main__":
    main()
tool_use

​
Reference

​
RespanTelemetry options

ArgumentDescription
app_nameApplication name shown in Respan.
api_keyOptional if RESPAN_API_KEY is set.
base_urlOptional; overrides RESPAN_BASE_URL.
is_enabledSet to False to disable tracing.
is_batching_enabledBatch export (default: typically True); set False for immediate flush in tests.

​
instrument_pydantic_ai() options

ArgumentDescription
agentOptional. If provided, only that agent is instrumented; if None, all agents are instrumented globally.
include_contentInclude message content in telemetry (default: True).
include_binary_contentInclude binary content in telemetry (default: True).

​
Instrument a single agent

By default, instrument_pydantic_ai() instruments all Pydantic AI agents globally. To instrument only one agent: 
agent = Agent("openai:gpt-4o")
instrument_pydantic_ai(agent=agent)

​
Further reading