> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
> Use this file to discover all available pages before exploring further.

# Schemas

> Pydantic schemas for agents, conversations, and API interactions in Swarms

## Overview

The `swarms.schemas` module provides Pydantic models and exception classes used for structured data validation across the framework. Everything importable from `swarms.schemas` falls into three groups:

| Group                      | Exports                                                                                                                                                                                               |
| -------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **MCP schemas**            | `MCPConnection`, `MCPOAuthConfig`                                                                                                                                                                     |
| **Agent errors**           | `AgentError`, `AgentInitializationError`, `AgentRunError`, `AgentLLMError`, `AgentLLMInitializationError`, `AgentToolExecutionError`, `AgentMCPError`, `AgentMCPConnectionError`, `AgentMCPToolError` |
| **Planner/worker schemas** | `TaskPriority`, `PlannerTaskStatus`, `PlannerTask`, `PlannerTaskOutput`, `PlannerTaskSpec`, `CycleVerdict`                                                                                            |

```python theme={null}
from swarms.schemas import (
    MCPConnection,
    MCPOAuthConfig,
    AgentError,
    PlannerTask,
    CycleVerdict,
)
```

<Note>
  These 17 names are the complete public surface of `swarms.schemas`.
</Note>

## MCP Schemas

### MCPConnection

Defines connection parameters for a Model Context Protocol (MCP) server.

```python theme={null}
from swarms.schemas import MCPConnection

mcp = MCPConnection(
    url="http://localhost:8000/mcp",
    name="filesystem",
    transport="streamable_http",
    authorization_token="secret-token",
    headers={"X-Custom-Header": "value"},
    timeout=30,
)
```

<ParamField path="type" type="str" default="mcp">
  The type of connection
</ParamField>

<ParamField path="url" type="str" default="http://localhost:8000/mcp">
  The URL endpoint for the MCP server
</ParamField>

<ParamField path="name" type="str" default="None">
  Human readable name for the server, used in logs and tool routing
</ParamField>

<ParamField path="tool_configurations" type="Dict[Any, Any]" default="None">
  Dictionary containing configuration settings for MCP tools
</ParamField>

<ParamField path="authorization_token" type="str" default="None">
  Bearer token for accessing the MCP server
</ParamField>

<ParamField path="api_key" type="str" default="None">
  API key for the MCP server. Sent using `api_key_header` / `api_key_prefix`
</ParamField>

<ParamField path="api_key_header" type="str" default="Authorization">
  Header used to send the API key, e.g. `"Authorization"` or `"X-API-Key"`
</ParamField>

<ParamField path="api_key_prefix" type="str" default="Bearer">
  Prefix prepended to the API key value. Set to `None`/`""` for raw keys
</ParamField>

<ParamField path="auth_type" type="Literal['none', 'api_key', 'bearer', 'oauth', 'custom']" default="None">
  Explicit auth mode. Inferred from the other fields when omitted
</ParamField>

<ParamField path="oauth" type="MCPOAuthConfig" default="None">
  OAuth 2.1 configuration for this server
</ParamField>

<ParamField path="transport" type="str" default="streamable_http">
  Transport protocol: `"streamable_http"`, `"sse"`, `"stdio"`, or `"auto"`
</ParamField>

<ParamField path="headers" type="Dict[str, str]" default="None">
  Headers to send to the MCP server
</ParamField>

<ParamField path="timeout" type="int" default="30">
  Request timeout (in seconds) for the MCP server
</ParamField>

<ParamField path="sse_read_timeout" type="int" default="300">
  How long to wait for streamed events before giving up
</ParamField>

<ParamField path="tool_timeout" type="int" default="120">
  How long a single tool call may run before timing out. Separate from `timeout`, which bounds HTTP requests
</ParamField>

<ParamField path="command" type="str" default="None">
  Executable to launch for the `stdio` transport
</ParamField>

<ParamField path="args" type="List[str]" default="None">
  Arguments passed to the `stdio` command
</ParamField>

<ParamField path="env" type="Dict[str, str]" default="None">
  Environment variables for the `stdio` command
</ParamField>

This model allows extra fields (`extra = "allow"`), so additional keys will not raise a validation error, but only the fields above are read by the framework.

### MCPOAuthConfig

OAuth 2.1 configuration for an MCP server, passed as `MCPConnection.oauth`. Three flavours are supported:

1. `grant_type="authorization_code"` (default) — the interactive browser flow. PKCE and RFC 7591 dynamic client registration are handled by the MCP SDK, so `client_id` is optional. Tokens are cached on disk so the browser prompt only happens once.
2. `grant_type="client_credentials"` — a headless machine-to-machine flow. Requires `client_id`/`client_secret`.
3. `access_token=...` — a token obtained elsewhere. No flow is run; the token is sent as a bearer credential.

Any string field may use `"env:MY_VAR"` or `"${MY_VAR}"` to read the value from the environment instead of hardcoding a secret.

```python theme={null}
from swarms.schemas import MCPConnection, MCPOAuthConfig

mcp = MCPConnection(
    url="https://mcp.example.com/mcp",
    oauth=MCPOAuthConfig(
        grant_type="client_credentials",
        client_id="env:MCP_CLIENT_ID",
        client_secret="env:MCP_CLIENT_SECRET",
        scopes=["mcp:tools"],
    ),
)
```

<ParamField path="grant_type" type="Literal['authorization_code', 'client_credentials']" default="authorization_code">
  OAuth grant to use when no static `access_token` is supplied
</ParamField>

<ParamField path="client_id" type="str" default="None">
  OAuth client id. Optional for `authorization_code` when the server supports dynamic client registration
</ParamField>

<ParamField path="client_secret" type="str" default="None">
  OAuth client secret. Required for `client_credentials`
</ParamField>

<ParamField path="scopes" type="List[str]" default="None">
  Scopes to request, e.g. `['mcp:tools', 'offline_access']`
</ParamField>

<ParamField path="redirect_uri" type="str" default="http://127.0.0.1:8765/callback">
  Loopback redirect URI used to capture the authorization code
</ParamField>

<ParamField path="client_name" type="str" default="Swarms Agent">
  Client name sent during dynamic client registration
</ParamField>

<ParamField path="client_uri" type="str" default="None">
  Client homepage sent during dynamic client registration
</ParamField>

<ParamField path="authorization_url" type="str" default="None">
  Explicit authorization endpoint. Discovered automatically when omitted
</ParamField>

<ParamField path="token_url" type="str" default="None">
  Explicit token endpoint. Discovered automatically when omitted
</ParamField>

<ParamField path="access_token" type="str" default="None">
  Pre-obtained access token. When set, no OAuth flow is performed
</ParamField>

<ParamField path="refresh_token" type="str" default="None">
  Pre-obtained refresh token, paired with `access_token`
</ParamField>

<ParamField path="token_storage_path" type="str" default="None">
  File used to cache OAuth tokens. Defaults to `~/.swarms/mcp_auth/<server>.json`
</ParamField>

<ParamField path="use_token_cache" type="bool" default="True">
  Persist tokens to disk so the browser flow is only run once
</ParamField>

<ParamField path="open_browser" type="bool" default="True">
  Open the system browser for the authorization step. When `False` the URL is logged instead
</ParamField>

<ParamField path="callback_timeout" type="int" default="300">
  Seconds to wait for the user to complete the browser flow
</ParamField>

## Agent Error Schemas

Exception classes raised by agents. All of them are plain Python exceptions — not Pydantic models — so they are caught with ordinary `try`/`except`.

Two independent hierarchies are exported:

```
AgentError                          AgentMCPError
├── AgentInitializationError        ├── AgentMCPConnectionError
├── AgentRunError                   └── AgentMCPToolError
├── AgentLLMError
├── AgentLLMInitializationError
└── AgentToolExecutionError
```

<Note>
  `AgentMCPError` inherits from `Exception`, **not** from `AgentError`. Catching `AgentError` will not catch MCP failures — catch both if you want blanket coverage.
</Note>

| Exception                     | Raised when                                                                        |
| ----------------------------- | ---------------------------------------------------------------------------------- |
| `AgentError`                  | Base class for all agent-related exceptions                                        |
| `AgentInitializationError`    | The agent fails to initialize; check configuration and parameters                  |
| `AgentRunError`               | The agent errors during execution; check the task and environment                  |
| `AgentLLMError`               | There is an issue with the language model; verify availability and compatibility   |
| `AgentLLMInitializationError` | The LLM fails to initialize; check configuration and parameters                    |
| `AgentToolExecutionError`     | The agent fails to execute a tool; check the tool's configuration and availability |
| `AgentMCPError`               | Base class for MCP-related failures                                                |
| `AgentMCPConnectionError`     | Connecting to an MCP server fails                                                  |
| `AgentMCPToolError`           | An MCP tool call fails                                                             |

```python theme={null}
from swarms import Agent
from swarms.schemas import (
    AgentError,
    AgentLLMError,
    AgentMCPConnectionError,
    AgentMCPError,
)

agent = Agent(agent_name="Analyst", model_name="gpt-5.4", max_loops=1)

try:
    result = agent.run("Summarise Q4 earnings.")
except AgentLLMError as e:
    print(f"Model problem: {e}")
except AgentMCPConnectionError as e:
    print(f"Could not reach the MCP server: {e}")
except (AgentError, AgentMCPError) as e:
    print(f"Agent failed: {e}")
```

## Planner/Worker Schemas

Schemas used by `PlannerWorkerSwarm` to represent tasks in the shared task queue.

### TaskPriority

`IntEnum` of task priority levels: `LOW = 0`, `NORMAL = 1`, `HIGH = 2`, `CRITICAL = 3`.

### PlannerTaskStatus

`str, Enum` of task statuses: `PENDING`, `CLAIMED`, `RUNNING`, `COMPLETED`, `FAILED`, `CANCELLED`.

Valid transitions:

```
PENDING -> CLAIMED -> RUNNING -> COMPLETED
PENDING -> CLAIMED -> RUNNING -> FAILED -> PENDING   (retry)
any non-terminal -> CANCELLED
```

### PlannerTask

A single task in the shared planner-worker task queue. Created by planner agents, consumed by worker agents. The `version` field enables optimistic concurrency control.

```python theme={null}
from swarms.schemas import PlannerTask, TaskPriority

task = PlannerTask(
    title="Research competitors",
    description="Gather competitor pricing data",
    priority=TaskPriority.HIGH,
)
```

<ParamField path="id" type="str" default="ptask-{uuid}">
  Unique task identifier
</ParamField>

<ParamField path="title" type="str" required>
  Short, descriptive title of the task
</ParamField>

<ParamField path="description" type="str" required>
  Detailed description of what needs to be done
</ParamField>

<ParamField path="priority" type="TaskPriority" default="NORMAL">
  Task priority level
</ParamField>

<ParamField path="depends_on" type="List[str]" default="[]">
  List of task IDs that must complete before this task can start
</ParamField>

<ParamField path="parent_task_id" type="str" default="None">
  ID of the parent task if decomposed from a larger task
</ParamField>

<ParamField path="status" type="PlannerTaskStatus" default="PENDING">
  Current task status
</ParamField>

<ParamField path="assigned_worker" type="str" default="None">
  Name of the worker agent that claimed this task
</ParamField>

<ParamField path="result" type="str" default="None">
  Result of task execution
</ParamField>

<ParamField path="error" type="str" default="None">
  Error message if task failed
</ParamField>

<ParamField path="retries" type="int" default="0">
  Number of retry attempts so far
</ParamField>

<ParamField path="max_retries" type="int" default="2">
  Maximum retry attempts before permanent failure
</ParamField>

<ParamField path="version" type="int" default="0">
  Optimistic concurrency version counter
</ParamField>

<ParamField path="created_at" type="float" default="time.time()">
  Unix timestamp of task creation
</ParamField>

<ParamField path="completed_at" type="float" default="None">
  Unix timestamp of task completion
</ParamField>

<ParamField path="metadata" type="Dict" default="{}">
  Arbitrary metadata
</ParamField>

### PlannerTaskOutput

A single task definition as output from a planner agent.

<ParamField path="title" type="str" required>
  Short, descriptive title
</ParamField>

<ParamField path="description" type="str" required>
  Detailed description of what a worker agent should do
</ParamField>

<ParamField path="priority" type="int" default="1">
  Priority: 0=LOW, 1=NORMAL, 2=HIGH, 3=CRITICAL
</ParamField>

<ParamField path="depends_on_titles" type="List[str]" default="[]">
  Titles of other tasks in this plan that must complete first
</ParamField>

### PlannerTaskSpec

Structured output from a planner agent: a plan narrative plus a list of concrete tasks.

```python theme={null}
from swarms.schemas import PlannerTaskOutput, PlannerTaskSpec

spec = PlannerTaskSpec(
    plan="Research the market, then summarise the findings.",
    tasks=[
        PlannerTaskOutput(
            title="Research competitors",
            description="Gather competitor pricing data",
            priority=2,
        ),
        PlannerTaskOutput(
            title="Write summary",
            description="Summarise the pricing research",
            depends_on_titles=["Research competitors"],
        ),
    ],
)
```

<ParamField path="plan" type="str" required>
  Narrative explanation of the plan: what needs to be done, in what order, and why
</ParamField>

<ParamField path="tasks" type="List[PlannerTaskOutput]" required>
  List of concrete tasks to add to the queue
</ParamField>

### CycleVerdict

Structured output from the judge agent after evaluating a planning cycle.

<ParamField path="is_complete" type="bool" required>
  True if the overall goal has been satisfactorily achieved
</ParamField>

<ParamField path="overall_quality" type="int" required>
  Quality score 0-10 of the combined results
</ParamField>

<ParamField path="summary" type="str" required>
  Summary assessment of the cycle results
</ParamField>

<ParamField path="gaps" type="List[str]" default="[]">
  Specific gaps or issues that need addressing in a follow-up cycle
</ParamField>

<ParamField path="follow_up_instructions" type="str" default="None">
  Instructions for the planner if another cycle is needed
</ParamField>

<ParamField path="needs_fresh_start" type="bool" default="False">
  True if accumulated drift or systemic issues require a complete restart rather than incremental gap-filling. When True, all prior tasks are discarded and the planner begins from scratch with the original goal plus judge feedback
</ParamField>

## Example: MCP Connection Setup

```python theme={null}
from swarms import Agent
from swarms.schemas import MCPConnection

# Describe a server with a full connection object
filesystem_mcp = MCPConnection(
    url="http://localhost:8000/mcp",
    name="filesystem",
    timeout=30,
)

agent = Agent(
    agent_name="MultiTool-Agent",
    mcp_url=filesystem_mcp,
    model_name="gpt-5.4",
)

# For multiple servers, Agent takes a list of URL strings via `mcp_urls`
# (not MCPConnection objects):
agent = Agent(
    agent_name="MultiServer-Agent",
    mcp_urls=["http://localhost:8000/mcp", "http://localhost:8001/mcp"],
    model_name="gpt-5.4",
)
```

## Example: Tracking Planner Tasks

```python theme={null}
import time

from swarms.schemas import PlannerTask, PlannerTaskStatus, TaskPriority

queue = [
    PlannerTask(
        title="Collect filings",
        description="Download the last four 10-Q filings",
        priority=TaskPriority.HIGH,
    ),
    PlannerTask(
        title="Summarise filings",
        description="Write a one-page summary of the filings",
    ),
]

# A worker claims and completes the first task
task = queue[0]
task.status = PlannerTaskStatus.RUNNING
task.assigned_worker = "Worker-1"
task.result = "Downloaded 4 filings."
task.status = PlannerTaskStatus.COMPLETED
task.completed_at = time.time()
task.version += 1

print(f"{task.id}: {task.status.value} by {task.assigned_worker}")
```

## Best Practices

1. **Type Safety**: Use the provided schemas for type-safe data handling
2. **Validation**: Leverage Pydantic's validation to catch errors early
3. **Serialization**: Use `.model_dump()` and `.model_dump_json()` for serialization
4. **Error Handling**: Catch the narrowest agent exception that fits, and remember `AgentError` and `AgentMCPError` are separate hierarchies
5. **MCP Configuration**: Use `MCPConnection` for consistent tool integration, and `MCPOAuthConfig` with `"env:VAR"` references instead of hardcoded secrets
6. **Concurrency**: Bump `PlannerTask.version` on every mutation so optimistic concurrency checks work

## Schema Inheritance

The Pydantic models above (`MCPConnection`, `MCPOAuthConfig`, `PlannerTask`, `PlannerTaskOutput`, `PlannerTaskSpec`, `CycleVerdict`) inherit from Pydantic's `BaseModel`, providing:

* Automatic validation
* JSON serialization/deserialization
* Schema generation
* IDE autocomplete support
* Type checking

```python theme={null}
from swarms.schemas import PlannerTask

task = PlannerTask(title="Research", description="Do research")

# Serialization
task_dict = task.model_dump()
task_json = task.model_dump_json(indent=2)

# Deserialization
task_from_dict = PlannerTask(**task_dict)
task_from_json = PlannerTask.model_validate_json(task_json)
```