pydanticai-platform 0.5.0
Multi-tenant platform for serving AI agents with PydanticAI
License: MIT
Requires Python: >=3.10
Keywords: agents, ai, anthropic, llm, multi-agent, multi-tenant, openai, platform, pydantic

# PydanticAI Multi-Tenant Agent Platform

A production-ready platform for serving AI agents to multiple tenants with [PydanticAI](https://github.com/pydantic/pydantic-ai).

## 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.

```python
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

```python
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:

```python
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:

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

## Adding Tools

```python
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

```python
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

```bash
# 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

```bash
# 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

```bash
# 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

```bash
# 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

```bash
# 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

```bash
# 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:

```bash
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

```python
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:

```env
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/](examples/starter/) | Simple agent, custom output, tools |
| [examples/multi_agent/](examples/multi_agent/) | Router + specialist delegation pattern |
| [examples/johnpye/](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](src/pydanticai_multiagent/api/app.py):

```python
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)](https://modelcontextprotocol.io/), 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`:

```bash
# Start the server
pydanticai-platform serve

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

### Stdio Transport (Local)

For Claude Desktop integration, run the MCP server directly:

```bash
pydanticai-platform mcp-serve
```

### Claude Desktop Configuration

Add to `claude_desktop_config.json`:

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

Or for a remote server:

```json
{
  "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:

```bash
# .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

```bash
# 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