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
)
step_id
str
default:"uuid.uuid4().hex"
Unique identifier for the task step
time
float
default:"current_time"
Time taken to complete the task step
response
AgentChatCompletionResponse
default:"None"
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
)
Unique identifier of the agent
Description of the task being executed
Maximum number of execution loops
run_id
str
default:"uuid.uuid4().hex"
Unique identifier for this execution run
steps
List[Union[Step, Any]]
default:"[]"
List of execution steps
Complete execution history as a string
Total number of tokens consumed
Token that caused execution to stop
Whether the task was interactive
dynamic_temperature_enabled
Whether dynamic temperature adjustment was enabled
MCP Schemas
MCPConnection
Defines connection parameters for Model Context Protocol (MCP) servers.
from swarms.schemas import MCPConnection
mcp = MCPConnection(
server_name="filesystem",
command="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/data"],
env={"API_KEY": "secret"}
)
MultipleMCPConnections
Manages multiple MCP server connections.
from swarms.schemas import MultipleMCPConnections
connections = MultipleMCPConnections(
connections=[mcp1, mcp2, mcp3]
)
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"
)
Object type (always “model”)
created
int
default:"current_timestamp"
Unix timestamp of model creation
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://..."}}
]
)
Role of message sender: ‘user’, ‘assistant’, or ‘system’
content
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
)
Model to use for completion
messages
List[ChatMessageInput]
required
List of messages in the conversation
Sampling temperature (0.0 to 2.0)
Nucleus sampling parameter
Maximum tokens to generate
Enable streaming responses
Penalty for token repetition
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
)
Model used for completion
object
Literal['chat.completion', 'chat.completion.chunk']
required
Response object type
choices
List[Union[ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice]]
required
List of completion choices
created
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
)
id
str
default:"agent-{uuid}"
Unique identifier for this agent response
Name of the agent that generated the response
object
Literal['chat.completion', 'chat.completion.chunk']
default:"None"
Response object type
choices
ChatCompletionResponseChoice
default:"None"
Completion choice data
created
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
)
Number of tokens in the prompt
Total tokens used (prompt + completion)
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(
server_name="filesystem",
command="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", "/data"]
)
web_search_mcp = MCPConnection(
server_name="brave-search",
command="npx",
args=["-y", "@modelcontextprotocol/server-brave-search"],
env={"BRAVE_API_KEY": "your-api-key"}
)
# Combine multiple connections
connections = MultipleMCPConnections(
connections=[filesystem_mcp, web_search_mcp]
)
# Use with agent
from swarms import Agent
agent = Agent(
agent_name="MultiTool-Agent",
mcp_connections=connections,
model_name="gpt-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)