Pydantic AI
Connect a Pydantic AI agent to Civic. For standalone scripts, use Pydantic AI's native MCPServerStreamableHTTP with a Civic token. For web apps, use per-user Civic Auth tokens.
Prerequisites
- Python 3.11+
- A Civic account at app.civic.com with a configured toolkit
Installation
pip install pydantic-ai python-dotenv
Authentication
For standalone scripts and autonomous agents — the simplest path.
Generate a Civic Token
- Log in to app.civic.com
- Click your account name in the bottom left
- Go to Install → MCP URL
- 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}` }
Token generation, URL parameters, OAuth vs token comparison
# .env
CIVIC_TOKEN=your-civic-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(
"https://app.civic.com/hub/mcp",
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())
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.
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}
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})
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 by adding profile to the MCP URL:
server = MCPServerStreamableHTTP(
f"https://app.civic.com/hub/mcp?profile={os.environ['CIVIC_PROFILE_ID']}",
headers={"Authorization": f"Bearer {os.environ['CIVIC_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
Complete implementation with FastAPI chat UI, streaming responses, and deployment guide