pydanticai-platform 0.5.0

Multi-tenant platform for serving AI agents with PydanticAI

PydanticAI Multi-Tenant Agent Platform

A production-ready platform for serving AI agents to multiple tenants with PydanticAI.

Who This Is For

Agencies and developers who build AI agents for clients.

You deploy one platform. Each client gets an API key. They call your agents via REST or SDK. You track usage and bill them.

Clients can: Chat with agents you've granted them. Continue conversations. See their usage.

Clients cannot: Create agents. Modify agents. See other tenants.

For Tenants

Tenants receive an API key from the platform admin.

pip install pydanticai-platform-client
from pydanticai_platform_client import PlatformClient

async with PlatformClient("https://platform.example.com", "pk_...") as client:
    response = await client.chat("Hello")
    print(response.text)

Full SDK docs: sdk/README.md

What This Is

This is a multi-tenant platform for serving AI agents. Deploy your agents once, then grant access to different clients/tenants via API keys.

Platform features:

• Multi-tenant architecture with isolated API keys
• Agent access control per tenant
• Conversation persistence per tenant
• Usage tracking and cost monitoring
• API key rotation
• Rate limiting per tenant

Infrastructure:

• FastAPI app with auth middleware, rate limiting, health checks
• Conversation persistence (Postgres/Supabase)
• Type-safe dependency injection
• Mock implementations for testing
• CLI for serving and chatting

You provide:

• Your agent(s) with domain-specific instructions
• Your tools for your use case
• Your output models

Quick Start

from pydantic_ai import Agent
from pydanticai_multiagent import BaseDeps, create_mock_base_deps

# 1. Define your agent
my_agent: Agent[BaseDeps, str] = Agent(
    "openai:gpt-4o",
    deps_type=BaseDeps,
    instructions="You are a helpful assistant for my specific domain...",
)

# 2. Run with mock deps (for development)
deps = create_mock_base_deps()
result = await my_agent.run("Hello!", deps=deps)
print(result.output)

Project Structure

src/pydanticai_multiagent/
├── dependencies/     # BaseDeps, SearchDeps, AuthDeps + mocks
├── tools/            # Reusable tools (search, data, external APIs)
├── services/         # Conversation, usage tracking
├── api/              # FastAPI app with middleware
├── db/               # Database pool + migrations
└── cli/              # Command-line interface

examples/
├── starter/          # Simple examples to learn the patterns
├── multi_agent/      # Full router + specialist agents example
└── johnpye/          # Real-world domain-specific example

Dependency Injection

Three levels of typed dependencies:

from pydanticai_multiagent import BaseDeps, SearchDeps, AuthDeps

# BaseDeps: Core infrastructure
#   - http_client, db, cache, user_id

# SearchDeps(BaseDeps): Adds search capabilities
#   - vector_store, search_api_key, domain_toolset

# AuthDeps(BaseDeps): Adds auth context
#   - user_roles, permissions

Convert between them when delegating:

# Your agent uses SearchDeps, sub-agent only needs BaseDeps
result = await sub_agent.run(query, deps=ctx.deps.to_base_deps())

Adding Tools

from pydantic_ai import RunContext
from pydantic_ai.toolsets import FunctionToolset
from pydanticai_multiagent import BaseDeps

my_toolset: FunctionToolset[BaseDeps] = FunctionToolset()

@my_toolset.tool
async def my_tool(ctx: RunContext[BaseDeps], query: str) -> str:
    """Do something useful."""
    # Access deps via ctx.deps.db, ctx.deps.http_client, etc.
    return f"Result for {query}"

# Use in your agent
my_agent = Agent(..., toolsets=[my_toolset])

Structured Outputs

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydanticai_multiagent import BaseDeps

class MyOutput(BaseModel):
    answer: str = Field(description="The answer")
    confidence: float = Field(ge=0, le=1)

my_agent: Agent[BaseDeps, MyOutput] = Agent(
    "openai:gpt-4o",
    deps_type=BaseDeps,
    output_type=MyOutput,
    instructions="...",
)

CLI

# Start the API server
pydanticai-platform serve
pydanticai-platform serve --reload  # with hot reload

# Interactive chat (uses example agents)
pydanticai-platform chat --agent router

# Single query
pydanticai-platform query "Your question" --agent research

# Run database migrations
pydanticai-platform db-migrate

# Run MCP server (for Claude Desktop, Cursor, etc.)
pydanticai-platform mcp-serve              # stdio transport (default)
pydanticai-platform mcp-serve --transport sse --port 8001  # SSE transport

Platform API

The platform provides two API layers: Admin API for tenant management and Platform API for tenant access.

Admin API (/api/v1/admin)

Requires admin authentication via Authorization: Bearer <SECRET_KEY>.

Tenant Management

# Create a tenant
curl -X POST http://localhost:8000/api/v1/admin/tenants \
  -H "Authorization: Bearer $SECRET_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name": "Acme Corp", "default_model": "openai:gpt-4o"}'

# Response includes API key (save it - only shown once):
# {"id": "tenant_xxx", "api_key": "sk-tenant-xxx", ...}

# List tenants
curl http://localhost:8000/api/v1/admin/tenants \
  -H "Authorization: Bearer $SECRET_KEY"

# Update tenant
curl -X PATCH "http://localhost:8000/api/v1/admin/tenants/{tenant_id}?is_active=false" \
  -H "Authorization: Bearer $SECRET_KEY"

# Rotate API key (invalidates old key immediately)
curl -X POST http://localhost:8000/api/v1/admin/tenants/{tenant_id}/rotate-key \
  -H "Authorization: Bearer $SECRET_KEY"

Agent Access Control

# Grant tenant access to an agent
curl -X POST http://localhost:8000/api/v1/admin/tenants/{tenant_id}/agents \
  -H "Authorization: Bearer $SECRET_KEY" \
  -H "Content-Type: application/json" \
  -d '{"agent_name": "support", "is_default": true}'

# List tenant's agents
curl http://localhost:8000/api/v1/admin/tenants/{tenant_id}/agents \
  -H "Authorization: Bearer $SECRET_KEY"

# Revoke agent access
curl -X DELETE http://localhost:8000/api/v1/admin/tenants/{tenant_id}/agents/support \
  -H "Authorization: Bearer $SECRET_KEY"

# List all platform agents
curl http://localhost:8000/api/v1/admin/agents \
  -H "Authorization: Bearer $SECRET_KEY"

Platform API (/api/v1/platform)

Requires tenant authentication via Authorization: Bearer <TENANT_API_KEY>.

Chat

# Send a message (uses default agent)
curl -X POST http://localhost:8000/api/v1/platform/chat \
  -H "Authorization: Bearer sk-tenant-xxx" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Hello!"}'

# Response:
# {"response": "Hi! How can I help?", "conversation_id": "conv_xxx", "agent": "support", "model": "openai:gpt-4o"}

# Continue conversation
curl -X POST http://localhost:8000/api/v1/platform/chat \
  -H "Authorization: Bearer sk-tenant-xxx" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Tell me more", "conversation_id": "conv_xxx"}'

# Use a specific agent
curl -X POST http://localhost:8000/api/v1/platform/chat \
  -H "Authorization: Bearer sk-tenant-xxx" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Analyze this data", "agent": "analyst"}'

# Streaming response (SSE)
curl -X POST http://localhost:8000/api/v1/platform/chat/stream \
  -H "Authorization: Bearer sk-tenant-xxx" \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Write a story"}'

Conversations

# List recent conversations
curl http://localhost:8000/api/v1/platform/conversations \
  -H "Authorization: Bearer sk-tenant-xxx"

# Get conversation history
curl http://localhost:8000/api/v1/platform/conversations/{conversation_id} \
  -H "Authorization: Bearer sk-tenant-xxx"

# Clear conversation
curl -X DELETE http://localhost:8000/api/v1/platform/conversations/{conversation_id} \
  -H "Authorization: Bearer sk-tenant-xxx"

Other Endpoints

# List available agents
curl http://localhost:8000/api/v1/platform/agents \
  -H "Authorization: Bearer sk-tenant-xxx"

# Get usage statistics
curl "http://localhost:8000/api/v1/platform/usage?days=30" \
  -H "Authorization: Bearer sk-tenant-xxx"

Database Setup

Run migrations to create the required tables:

pydanticai-multiagent db-migrate

Required tables:

• tenants - Tenant accounts with API key hashes
• agents - Registered agents
• tenant_agents - Agent access per tenant
• conversation_messages - Conversation history
• usage_records - Usage tracking

Testing

from pydantic_ai.models.test import TestModel
from pydanticai_multiagent import create_mock_base_deps

async def test_my_agent():
    deps = create_mock_base_deps()

    with my_agent.override(model=TestModel()):
        result = await my_agent.run("test query", deps=deps)
        assert "expected" in result.output

Configuration

Create a .env file:

OPENAI_API_KEY=sk-...
DATABASE_URL=postgresql://...  # Optional, uses mocks if not set
SECRET_KEY=change-me           # For auth middleware

Examples

Example	Description
examples/starter/	Simple agent, custom output, tools
examples/multi_agent/	Router + specialist delegation pattern
examples/johnpye/	Real-world auction tracking agent

Built-in Agents

The platform includes example agents you can grant to tenants:

Agent	Description
router	Routes requests to specialist agents
research	Web research and information gathering
analyst	Data analysis and insights
code	Code generation and review
writer	Content writing and editing
support	Customer support assistance

Register your own agents in app.py:

registry.register_builtin("my_agent", my_agent, "Description")

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Admin API                                │
│  POST /admin/tenants    - Create tenant (returns API key)   │
│  POST /admin/tenants/{id}/agents - Grant agent access       │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Tenant Database                           │
│  tenants (id, api_key_hash, default_model, rate_limit)      │
│  tenant_agents (tenant_id, agent_id, model_override)        │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Platform API                              │
│  POST /platform/chat - Send message (tenant API key auth)   │
│  GET /platform/conversations - List tenant's conversations  │
└─────────────────────────────────────────────────────────────┘

Platform Status

Production-Ready

Feature	Implementation
Tenant API keys	SHA256 hashed, secure storage
Tenant CRUD	Full create/read/update via Admin API
Agent registry	Database-backed with foreign key integrity
Agent access control	Per-tenant permissions via tenant_agents
Conversation persistence	PostgreSQL JSONB with PydanticAI serialization
API key rotation	Immediate invalidation of old keys
Database migrations	Tracked, idempotent migrations
Usage tracking	ConversationService calls record_usage() after each chat
Token limit enforcement	monthly_token_limit checked before chat, returns 429 when exceeded
Cost monitoring	Token usage tracked per tenant via /usage endpoint
MCP server	Exposes platform as MCP server for Claude Desktop, Cursor, etc.
Platform-wide MCP servers	External MCP servers available to all agents via MCP_SERVERS config

Partial Implementation

Feature	Status	Notes
Rate limiting	In-memory, per-tenant	Works for single instance. For multi-instance deployments, implement Redis-based rate limiting.

To Reach Full Production

1. Distributed rate limiting - Replace in-memory rate limiter with Redis for multi-instance deployments

MCP Integration

The platform can be accessed via Model Context Protocol (MCP), enabling Claude Desktop, Cursor, and other MCP clients to use your agents.

Available MCP Tools

Tool	Description
chat	Send a message to an agent
list_agents	List available agents
get_usage	Get usage statistics
list_conversations	List recent conversations
get_conversation	Get conversation history
clear_conversation	Clear a conversation

Each tool requires an api_key parameter for authentication (your tenant API key).

HTTP Transport (Remote Server)

When running the API server, MCP is available at /mcp:

# Start the server
pydanticai-platform serve

# MCP endpoint: http://localhost:8000/mcp

Stdio Transport (Local)

For Claude Desktop integration, run the MCP server directly:

pydanticai-platform mcp-serve

Claude Desktop Configuration

Add to claude_desktop_config.json:

{
  "mcpServers": {
    "pydanticai-platform": {
      "command": "pydanticai-platform",
      "args": ["mcp-serve"]
    }
  }
}

Or for a remote server:

{
  "mcpServers": {
    "pydanticai-platform": {
      "url": "https://your-platform.fly.dev/mcp",
      "transport": "streamable-http"
    }
  }
}

Example Usage

Once configured, you can ask Claude:

"Use the pydanticai-platform to chat with the research agent about climate change. My API key is sk-tenant-xxx."

Claude will call the chat tool with your API key and prompt, and return the agent's response.

Platform-wide MCP Servers

The platform can connect to external MCP servers, making their tools available to all agents. This allows your agents to use tools from the MCP ecosystem (time, fetch, filesystem, databases, etc.).

Configuration

Set the MCP_SERVERS environment variable with a JSON array of server configs:

# .env
MCP_SERVERS='[
  {"type": "stdio", "command": "uvx", "args": ["mcp-server-time"], "prefix": "time"},
  {"type": "stdio", "command": "npx", "args": ["-y", "@anthropics/mcp-server-fetch"], "prefix": "fetch"},
  {"type": "http", "url": "http://localhost:8080/mcp", "prefix": "custom"}
]'

Supported Transport Types

Type	Description	Required Fields
stdio	Local command-based servers	command, optionally args, env
sse	Remote SSE servers (deprecated)	url, optionally headers
http	Remote HTTP servers (Streamable HTTP)	url, optionally headers

Configuration Fields

Field	Type	Description
type	string	Transport type: stdio, sse, or http
prefix	string	Tool name prefix (e.g., time makes time_get_current_time)
timeout	int	Request timeout in seconds (default: 30)
command	string	Command to run (stdio only)
args	list	Command arguments (stdio only)
env	dict	Environment variables (stdio only)
url	string	Server URL (sse/http only)
headers	dict	HTTP headers (sse/http only)

Example: Adding Time Tools

# Install the MCP server
uvx mcp-server-time --help

# Configure in .env
MCP_SERVERS='[{"type": "stdio", "command": "uvx", "args": ["mcp-server-time"], "prefix": "time"}]'

Now all agents can use time_get_current_time and other time-related tools.

What's NOT Included

• Vendor lock-in (swap models, databases, etc.)
• Tenant-defined agents (agents are server-defined for security)

Define your agents in code, grant access to tenants via Admin API.

License

MIT