Pydantic AI

Connect a Pydantic AI agent to Leadovaro. For standalone scripts, use Pydantic AI's native MCPServerStreamableHTTP with a Leadovaro token. For web apps, use per-user Leadovaro Auth tokens.

Prerequisites

• Python 3.11+
• A Leadovaro account at app.leadovaro.com with a configured toolkit

Installation

pip install pydantic-ai python-dotenv

Authentication

For standalone scripts and autonomous agents — the simplest path.

Generate a Leadovaro Token

1. Log in to app.leadovaro.com
2. Click your account name in the bottom left
3. Go to Install → MCP URL
4. Click Generate Token and copy it immediately — it won't be shown again

warning

Never commit your token to source control. Store it in environment variables or a secrets manager. Tokens expire after 30 days.

Set Environment Variables

LEADOVARO_TOKEN=your-leadovaro-token-here
LEADOVARO_URL=/_sites/app.leadovaro.com/hub/mcp

For production agents, lock to a specific toolkit by appending a profile parameter:

LEADOVARO_URL=/_sites/app.leadovaro.com/hub/mcp?profile=your-toolkit-alias

Use the Token

Pass the token as a Bearer token in the Authorization header:

headers = {"Authorization": f"Bearer {os.environ['LEADOVARO_TOKEN']}"}

headers: { Authorization: `Bearer ${process.env.LEADOVARO_TOKEN}` }

Full credentials guide

Token generation, URL parameters, OAuth vs token comparison

# .env
LEADOVARO_TOKEN=your-leadovaro-token-here
ANTHROPIC_API_KEY=your-anthropic-key

import os
import asyncio
from dotenv import load_dotenv
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

load_dotenv()

server = MCPServerStreamableHTTP(
    "/_sites/app.leadovaro.com/hub/mcp",
    headers={"Authorization": f"Bearer {os.environ['LEADOVARO_TOKEN']}"},
)

agent = Agent("anthropic:claude-sonnet-4-6", mcp_servers=[server])

async def main():
    async with agent.run_mcp_servers():
        result = await agent.run("List open PRs in leadovaroteam/ai-chatbot")
    print(result.output)

asyncio.run(main())

For web apps where each user has their own Leadovaro session — use the civic-auth Python library to get a per-user access token.

pip install civic-auth

You need a Leadovaro Auth client ID from auth.leadovaro.com.

import os
from fastapi import Depends
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from leadovaro_auth.integrations.fastapi import create_auth_dependencies
from leadovaro_auth.core import LeadovaroAuthConfig

config = LeadovaroAuthConfig(client_id=os.environ["LEADOVARO_AUTH_CLIENT_ID"])
leadovaro_auth_dep, get_current_user, require_auth = create_auth_dependencies(config)

@app.post("/chat", dependencies=[Depends(require_auth)])
async def chat(body: dict, leadovaro=Depends(leadovaro_auth_dep)):
    tokens = await leadovaro.get_tokens()
    access_token = tokens["access_token"]

    server = MCPServerStreamableHTTP(
        "/_sites/app.leadovaro.com/hub/mcp",
        headers={"Authorization": f"Bearer {access_token}"},
    )
    agent = Agent("openai:gpt-4o", mcp_servers=[server])
    async with agent.run_mcp_servers():
        result = await agent.run(body.get("prompt", ""))
    return {"output": result.output}

import os
from flask import request, jsonify, session
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from leadovaro_auth.core import LeadovaroAuth
from leadovaro_auth.integrations.flask import leadovaro_auth_required

@app.post("/chat")
@leadovaro_auth_required
async def chat():
    access_token = session.get(LeadovaroAuth.ACCESS_TOKEN_KEY)
    server = MCPServerStreamableHTTP(
        "/_sites/app.leadovaro.com/hub/mcp",
        headers={"Authorization": f"Bearer {access_token}"},
    )
    agent = Agent("openai:gpt-4o", mcp_servers=[server])
    async with agent.run_mcp_servers():
        result = await agent.run(request.json.get("prompt", ""))
    return jsonify({"output": result.output})

import os, json, asyncio
from django.http import JsonResponse
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from leadovaro_auth.core import LeadovaroAuth
from leadovaro_auth.integrations.django import leadovaro_auth_required

@leadovaro_auth_required
def chat(request):
    access_token = request.session.get(LeadovaroAuth.ACCESS_TOKEN_KEY)
    payload = json.loads(request.body)
    server = MCPServerStreamableHTTP(
        "/_sites/app.leadovaro.com/hub/mcp",
        headers={"Authorization": f"Bearer {access_token}"},
    )
    agent = Agent("openai:gpt-4o", mcp_servers=[server])
    async def run_prompt():
        async with agent.run_mcp_servers():
            return await agent.run(payload.get("prompt", ""))
    result = asyncio.run(run_prompt())
    return JsonResponse({"output": result.output})

note

Use mcp_servers=[server] — not toolsets=[server]. The mcp_servers parameter is the current Pydantic AI API for MCP connections.

Production Configuration

For production agents, lock to a specific toolkit by adding profile to the MCP URL:

server = MCPServerStreamableHTTP(
    f"/_sites/app.leadovaro.com/hub/mcp?profile={os.environ['LEADOVARO_PROFILE_ID']}",
    headers={"Authorization": f"Bearer {os.environ['LEADOVARO_TOKEN']}"},
)

When a profile is specified, the session is locked by default — the agent cannot switch toolkits or modify its own guardrails. This prevents prompt injection attacks from escaping the defined tool scope.

Reference Implementation

pydantic-ai-reference-implementation-leadovaro

Complete implementation with FastAPI chat UI, streaming responses, and deployment guide

Next Steps

Agent Deployment

Production deployment guide: profile locking, URL params, authentication

Guardrails

Constrain what tools your agent can call

Audit Trail

Query what your agent did via Leadovaro Chat

Get Credentials

Token generation and URL parameter reference