Pydantic AI

Connect a Pydantic AI agent to Civic’s MCP Hub using streamable HTTP transport

Connect a Pydantic AI agent to Civic using native MCP support. Pydantic AI’s MCPServerStreamableHTTP class handles session management and tool discovery automatically.

Prerequisites

• Python 3.11+
• A Civic account at app.civic.com with a configured toolkit

Installation

pip install pydantic-ai python-dotenv

Authentication

Script / Agent (Civic Token)

For standalone scripts and autonomous agents — the simplest path.

Generate a Civic Token

1. Log in to app.civic.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

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

Set Environment Variables

CIVIC_TOKEN=your-civic-token-here
CIVIC_URL=https://app.civic.com/hub/mcp

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

CIVIC_URL=https://app.civic.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['CIVIC_TOKEN']}"}

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

Full credentials guide

Token generation, URL parameters, OAuth vs token comparison

# .env
CIVIC_TOKEN=your-civic-token-here
CIVIC_URL=https://app.civic.com/hub/mcp

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

load_dotenv()

server = MCPServerStreamableHTTP(
    os.environ["CIVIC_URL"],
    headers={"Authorization": f"Bearer {os.environ['CIVIC_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 civicteam/ai-chatbot")
    print(result.output)

asyncio.run(main())

You can customize tool names, arguments, and filtering. See Pydantic AI MCP client docs.

Web App (Civic Auth)

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

pip install civic-auth

You need a Civic Auth client ID from auth.civic.com.

FastAPI

import os
from fastapi import Depends
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from civic_auth.integrations.fastapi import create_auth_dependencies
from civic_auth.core import CivicAuthConfig

config = CivicAuthConfig(client_id=os.environ["CIVIC_AUTH_CLIENT_ID"])
civic_auth_dep, get_current_user, require_auth = create_auth_dependencies(config)

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

    server = MCPServerStreamableHTTP(
        "https://app.civic.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}

Flask

import os
from flask import request, jsonify, session
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from civic_auth.core import CivicAuth
from civic_auth.integrations.flask import civic_auth_required

@app.post("/chat")
@civic_auth_required
async def chat():
    access_token = session.get(CivicAuth.ACCESS_TOKEN_KEY)
    server = MCPServerStreamableHTTP(
        "https://app.civic.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})

Django

import os, json, asyncio
from django.http import JsonResponse
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from civic_auth.core import CivicAuth
from civic_auth.integrations.django import civic_auth_required

@civic_auth_required
def chat(request):
    access_token = request.session.get(CivicAuth.ACCESS_TOKEN_KEY)
    payload = json.loads(request.body)
    server = MCPServerStreamableHTTP(
        "https://app.civic.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})

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 using the profile URL parameter:

CIVIC_URL=https://app.civic.com/hub/mcp?profile=your-production-toolkit

When a profile is specified, the session is locked by default — the agent cannot switch toolkits or modify its own guardrails.

Reference Implementation

pydantic-ai-reference-implementation-civic

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 Civic Chat

Get Credentials

Token generation and URL parameter reference