Schemas

Overview

The swarms.schemas module provides Pydantic BaseModel schemas for structured data validation, API interactions, and agent step tracking. These schemas ensure type safety and data consistency across the Swarms framework.

Agent Step Schemas

Step

Represents a single execution step in an agent’s workflow.

from swarms.schemas import Step

step = Step(
    step_id="unique-step-id",
    time=1.234,
    response=agent_response
)

str

default:"uuid.uuid4().hex"

Unique identifier for the task step

float

default:"current_time"

Time taken to complete the task step

AgentChatCompletionResponse

required

Agent’s response for this step

ManySteps

Tracks multiple execution steps and agent run metadata.

from swarms.schemas import ManySteps

run_data = ManySteps(
    agent_id="financial-agent-1",
    agent_name="FinancialAnalyst",
    task="Analyze Q4 earnings",
    max_loops=3,
    steps=[step1, step2, step3],
    total_tokens=5000
)

str

required

Unique identifier of the agent

str

required

Name of the agent

str

required

Description of the task being executed

Any

required

Maximum number of execution loops

str

default:"uuid.uuid4().hex"

Unique identifier for this execution run

List[Union[Step, Any]]

default:"[]"

List of execution steps

str

required

Complete execution history as a string

int

required

Total number of tokens consumed

str

required

Token that caused execution to stop

bool

required

Whether the task was interactive

bool

required

Whether dynamic temperature adjustment was enabled

MCP Schemas

MCPConnection

Defines connection parameters for a Model Context Protocol (MCP) server reachable over HTTP/SSE.

from swarms.schemas import MCPConnection

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

str

default:"mcp"

The type of connection

str

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

The URL endpoint for the MCP server

Dict[Any, Any]

default:"None"

Dictionary containing configuration settings for MCP tools

str

default:"None"

Authentication token for accessing the MCP server

str

default:"streamable_http"

The transport protocol to use for the MCP server

Dict[str, str]

default:"None"

Headers to send to the MCP server

int

default:"10"

Timeout (in seconds) for the MCP server

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.

MultipleMCPConnections

Manages multiple MCP server connections.

from swarms.schemas import MultipleMCPConnections

connections = MultipleMCPConnections(
    connections=[mcp1, mcp2, mcp3]
)

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.

PlannerTask

A single task in the shared planner-worker task queue.

from swarms.schemas import PlannerTask

task = PlannerTask(
    title="Research competitors",
    description="Gather competitor pricing data",
)

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

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.

str

required

Narrative explanation of the plan

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 requires discarding all prior tasks and restarting from the original goal

Base Schemas

ModelCard

Metadata about a machine learning model.

from swarms.schemas.base_schemas import ModelCard

model = ModelCard(
    id="gpt-4",
    object="model",
    created=1234567890,
    owned_by="openai"
)

str

required

Model identifier

str

default:"model"

Object type (always “model”)

int

default:"current_timestamp"

Unix timestamp of model creation

str

default:"owner"

Model owner/organization

str

default:"None"

Root model identifier

str

default:"None"

Parent model identifier

list

default:"None"

Model permissions

ChatMessageInput

Input message for chat completions.

from swarms.schemas.base_schemas import ChatMessageInput

message = ChatMessageInput(
    role="user",
    content="Analyze this financial report"
)

# Multi-modal message
multimodal_message = ChatMessageInput(
    role="user",
    content=[
        {"type": "text", "text": "What's in this image?"},
        {"type": "image_url", "image_url": {"url": "https://..."}}
    ]
)

str

required

Role of message sender: ‘user’, ‘assistant’, or ‘system’

Union[str, List[ContentItem]]

required

Message content (text or multi-modal)

ChatCompletionRequest

Request schema for chat completions.

from swarms.schemas.base_schemas import ChatCompletionRequest

request = ChatCompletionRequest(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"}
    ],
    temperature=0.7,
    max_tokens=1000,
    stream=False
)

str

default:"gpt-5.4"

Model to use for completion

List[ChatMessageInput]

required

List of messages in the conversation

float

default:"0.8"

Sampling temperature (0.0 to 2.0)

float

default:"0.8"

Nucleus sampling parameter

int

default:"4000"

Maximum tokens to generate

bool

default:"False"

Enable streaming responses

float

default:"1.0"

Penalty for token repetition

bool

default:"False"

Whether to echo the prompt in addition to the completion

ChatCompletionResponse

Response schema for chat completions.

from swarms.schemas.base_schemas import ChatCompletionResponse

response = ChatCompletionResponse(
    model="gpt-4",
    object="chat.completion",
    choices=[
        {
            "index": 0,
            "message": {"role": "assistant", "content": "Hello! How can I help?"},
            "input": "Hello!"
        }
    ],
    created=1234567890
)

str

required

Model used for completion

Literal['chat.completion', 'chat.completion.chunk']

required

Response object type

List[Union[ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice]]

required

List of completion choices

int

default:"current_timestamp"

Unix timestamp of response creation

AgentChatCompletionResponse

Agent-specific chat completion response with tracking metadata.

from swarms.schemas.base_schemas import AgentChatCompletionResponse

response = AgentChatCompletionResponse(
    id="agent-abc123",
    agent_name="FinancialAnalyst",
    object="chat.completion",
    choices=choice_data,
    created=1234567890
)

str

default:"agent-{uuid}"

Unique identifier for this agent response

str

required

Name of the agent that generated the response

Literal['chat.completion', 'chat.completion.chunk']

default:"None"

Response object type

ChatCompletionResponseChoice

default:"None"

Completion choice data

int

default:"current_timestamp"

Unix timestamp of response creation

UsageInfo

Token usage information for API calls.

from swarms.schemas.base_schemas import UsageInfo

usage = UsageInfo(
    prompt_tokens=150,
    completion_tokens=300,
    total_tokens=450
)

int

default:"0"

Number of tokens in the prompt

int

default:"0"

Total tokens used (prompt + completion)

int

default:"0"

Number of tokens in the completion

Example: Complete Agent Tracking

from swarms.schemas import Step, ManySteps
from swarms.schemas.base_schemas import AgentChatCompletionResponse
import time

# Track individual steps
steps = []

for i in range(3):
    step_start = time.time()
    
    # Simulate agent execution
    response = AgentChatCompletionResponse(
        agent_name="FinancialAnalyst",
        choices={"message": {"role": "assistant", "content": f"Analysis {i+1}"}}
    )
    
    step = Step(
        step_id=f"step-{i}",
        time=time.time() - step_start,
        response=response
    )
    steps.append(step)

# Create complete run record
run_record = ManySteps(
    agent_id="financial-agent-1",
    agent_name="FinancialAnalyst",
    task="Analyze Q4 earnings reports",
    max_loops=3,
    run_id="run-abc123",
    steps=steps,
    full_history="Complete execution history...",
    total_tokens=5000,
    stopping_token="<END>",
    interactive=False,
    dynamic_temperature_enabled=True
)

print(f"Run ID: {run_record.run_id}")
print(f"Total steps: {len(run_record.steps)}")
print(f"Total tokens: {run_record.total_tokens}")

Example: MCP Connection Setup

from swarms.schemas import MCPConnection, MultipleMCPConnections

# Define individual connections
filesystem_mcp = MCPConnection(
    url="http://localhost:8000/sse",
)

web_search_mcp = MCPConnection(
    url="http://localhost:8001/sse",
    authorization_token="your-api-key",
)

# Combine multiple connections
connections = MultipleMCPConnections(
    connections=[filesystem_mcp, web_search_mcp]
)

# Use a single connection with an agent
from swarms import Agent

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/sse", "http://localhost:8001/sse"],
    model_name="gpt-5.4",
)

Best Practices

Type Safety: Always 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
Step Tracking: Track all agent steps for debugging and analysis
Token Monitoring: Monitor token usage through UsageInfo schemas
MCP Configuration: Use MCPConnection schemas for consistent tool integration

Schema Inheritance

All schemas inherit from Pydantic’s BaseModel, providing:

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

# Serialization
step_dict = step.model_dump()
step_json = step.model_dump_json(indent=2)

# Deserialization
step_from_dict = Step(**step_dict)
step_from_json = Step.model_validate_json(step_json)

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

Agent Step Schemas

Step

ManySteps

MCP Schemas

MCPConnection

MultipleMCPConnections

Planner/Worker Schemas

TaskPriority

PlannerTaskStatus

PlannerTask

PlannerTaskOutput

PlannerTaskSpec

CycleVerdict

Base Schemas

ModelCard

ChatMessageInput

ChatCompletionRequest

ChatCompletionResponse

AgentChatCompletionResponse

UsageInfo

Example: Complete Agent Tracking

Example: MCP Connection Setup

Best Practices

Schema Inheritance

​Overview

​Agent Step Schemas

​Step

​ManySteps

​MCP Schemas

​MCPConnection

​MultipleMCPConnections

​Planner/Worker Schemas

​TaskPriority

​PlannerTaskStatus

​PlannerTask

​PlannerTaskOutput

​PlannerTaskSpec

​CycleVerdict

​Base Schemas

​ModelCard

​ChatMessageInput

​ChatCompletionRequest

​ChatCompletionResponse

​AgentChatCompletionResponse

​UsageInfo

​Example: Complete Agent Tracking

​Example: MCP Connection Setup

​Best Practices

​Schema Inheritance

Overview

Agent Step Schemas

Step

ManySteps

MCP Schemas

MCPConnection

MultipleMCPConnections

Planner/Worker Schemas

TaskPriority

PlannerTaskStatus

PlannerTask

PlannerTaskOutput

PlannerTaskSpec

CycleVerdict

Base Schemas

ModelCard

ChatMessageInput

ChatCompletionRequest

ChatCompletionResponse

AgentChatCompletionResponse

UsageInfo

Example: Complete Agent Tracking

Example: MCP Connection Setup

Best Practices

Schema Inheritance