Schemas

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`

from swarms.schemas import (
    MCPConnection,
    MCPOAuthConfig,
    AgentError,
    PlannerTask,
    CycleVerdict,
)

These 17 names are the complete public surface of swarms.schemas.

MCP Schemas

MCPConnection

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

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,
)

str

default:"mcp"

The type of connection

str

default:"http://localhost:8000/mcp"

The URL endpoint for the MCP server

str

default:"None"

Human readable name for the server, used in logs and tool routing

Dict[Any, Any]

default:"None"

Dictionary containing configuration settings for MCP tools

str

default:"None"

Bearer token for accessing the MCP server

str

default:"None"

API key for the MCP server. Sent using api_key_header / api_key_prefix

str

default:"Authorization"

Header used to send the API key, e.g. "Authorization" or "X-API-Key"

str

default:"Bearer"

Prefix prepended to the API key value. Set to None/"" for raw keys

Literal['none', 'api_key', 'bearer', 'oauth', 'custom']

default:"None"

Explicit auth mode. Inferred from the other fields when omitted

MCPOAuthConfig

default:"None"

OAuth 2.1 configuration for this server

str

default:"streamable_http"

Transport protocol: "streamable_http", "sse", "stdio", or "auto"

Dict[str, str]

default:"None"

Headers to send to the MCP server

int

default:"30"

Request timeout (in seconds) for the MCP server

int

default:"300"

How long to wait for streamed events before giving up

int

default:"120"

How long a single tool call may run before timing out. Separate from timeout, which bounds HTTP requests

str

default:"None"

Executable to launch for the stdio transport

List[str]

default:"None"

Arguments passed to the stdio command

Dict[str, str]

default:"None"

Environment variables for the stdio command

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:

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.
grant_type="client_credentials" — a headless machine-to-machine flow. Requires client_id/client_secret.
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.

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"],
    ),
)

Literal['authorization_code', 'client_credentials']

default:"authorization_code"

OAuth grant to use when no static access_token is supplied

str

default:"None"

OAuth client id. Optional for authorization_code when the server supports dynamic client registration

str

default:"None"

OAuth client secret. Required for client_credentials

List[str]

default:"None"

Scopes to request, e.g. ['mcp:tools', 'offline_access']

str

default:"http://127.0.0.1:8765/callback"

Loopback redirect URI used to capture the authorization code

str

default:"Swarms Agent"

Client name sent during dynamic client registration

str

default:"None"

Client homepage sent during dynamic client registration

str

default:"None"

Explicit authorization endpoint. Discovered automatically when omitted

str

default:"None"

Explicit token endpoint. Discovered automatically when omitted

str

default:"None"

Pre-obtained access token. When set, no OAuth flow is performed

str

default:"None"

Pre-obtained refresh token, paired with access_token

str

default:"None"

File used to cache OAuth tokens. Defaults to ~/.swarms/mcp_auth/<server>.json

bool

default:"True"

Persist tokens to disk so the browser flow is only run once

bool

default:"True"

Open the system browser for the authorization step. When False the URL is logged instead

int

default:"300"

Seconds to wait for the user to complete the browser flow

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

AgentMCPError inherits from Exception, not from AgentError. Catching AgentError will not catch MCP failures — catch both if you want blanket coverage.

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

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.

from swarms.schemas import PlannerTask, TaskPriority

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

str

default:"ptask-{uuid}"

Unique task identifier

str

required

Short, descriptive title of the task

str

required

Detailed description of what needs to be done

TaskPriority

default:"NORMAL"

Task priority level

List[str]

default:"[]"

List of task IDs that must complete before this task can start

str

default:"None"

ID of the parent task if decomposed from a larger task

PlannerTaskStatus

default:"PENDING"

Current task status

str

default:"None"

Name of the worker agent that claimed this task

str

default:"None"

Result of task execution

str

default:"None"

Error message if task failed

int

default:"0"

Number of retry attempts so far

int

default:"2"

Maximum retry attempts before permanent failure

int

default:"0"

Optimistic concurrency version counter

float

default:"time.time()"

Unix timestamp of task creation

float

default:"None"

Unix timestamp of task completion

Dict

default:"{}"

Arbitrary metadata

PlannerTaskOutput

A single task definition as output from a planner agent.

str

required

Short, descriptive title

str

required

Detailed description of what a worker agent should do

int

default:"1"

Priority: 0=LOW, 1=NORMAL, 2=HIGH, 3=CRITICAL

List[str]

default:"[]"

Titles of other tasks in this plan that must complete first

PlannerTaskSpec

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

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"],
        ),
    ],
)

str

required

Narrative explanation of the plan: what needs to be done, in what order, and why

List[PlannerTaskOutput]

required

List of concrete tasks to add to the queue

CycleVerdict

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

bool

required

True if the overall goal has been satisfactorily achieved

int

required

Quality score 0-10 of the combined results

str

required

Summary assessment of the cycle results

List[str]

default:"[]"

Specific gaps or issues that need addressing in a follow-up cycle

str

default:"None"

Instructions for the planner if another cycle is needed

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

Example: MCP Connection Setup

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

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

Type Safety: Use the provided schemas for type-safe data handling
Validation: Leverage Pydantic’s validation to catch errors early
Serialization: Use .model_dump() and .model_dump_json() for serialization
Error Handling: Catch the narrowest agent exception that fits, and remember AgentError and AgentMCPError are separate hierarchies
MCP Configuration: Use MCPConnection for consistent tool integration, and MCPOAuthConfig with "env:VAR" references instead of hardcoded secrets
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

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)

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

MCP Schemas

MCPConnection

MCPOAuthConfig

Agent Error Schemas

Planner/Worker Schemas

TaskPriority

PlannerTaskStatus

PlannerTask

PlannerTaskOutput

PlannerTaskSpec

CycleVerdict

Example: MCP Connection Setup

Example: Tracking Planner Tasks

Best Practices

Schema Inheritance

​Overview

​MCP Schemas

​MCPConnection

​MCPOAuthConfig

​Agent Error Schemas

​Planner/Worker Schemas

​TaskPriority

​PlannerTaskStatus

​PlannerTask

​PlannerTaskOutput

​PlannerTaskSpec

​CycleVerdict

​Example: MCP Connection Setup

​Example: Tracking Planner Tasks

​Best Practices

​Schema Inheritance

Overview

MCP Schemas

MCPConnection

MCPOAuthConfig

Agent Error Schemas

Planner/Worker Schemas

TaskPriority

PlannerTaskStatus

PlannerTask

PlannerTaskOutput

PlannerTaskSpec

CycleVerdict

Example: MCP Connection Setup

Example: Tracking Planner Tasks

Best Practices

Schema Inheritance