pydanticai-platform 0.3.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-multiagent serve
pydanticai-multiagent serve --reload  # with hot reload

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

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

# Run database migrations
pydanticai-multiagent db-migrate

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

Partial Implementation

Feature	Status	Notes
Rate limiting	In-memory only	Works for single instance. For multi-instance deployments, implement Redis-based rate limiting.
Per-tenant rate limits	Config stored, not enforced	rate_limit_per_minute field exists on Tenant but rate limiter uses global config.

To Reach Production

1. Enforce tenant rate limits - Read tenant.rate_limit_per_minute in rate limit middleware
2. Distributed rate limiting - Replace in-memory rate limiter with Redis for multi-instance deployments

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