Migration Guide: v0.x to v1.x

​

Installation

​

Agency Swarm v1.x (Beta)

pip install -U "agency-swarm>=1.0.0-beta.1"

​

Agency Swarm v0.x (Production)

pip install "agency-swarm<1.0.0"

​

New Features & Capabilities

​

Responses API Integration & New Models

• Web Search & Computer Use: Native support for OpenAI’s Search and Computer Use capabilities through the Responses API
• Latest Models: Full support for o3, o4-mini, and future OpenAI models via Responses API
• Third-Party Model Providers: Use any model provider that adheres to the Chat Completions API standard
• Direct Control: No more black-box Assistants API - Agency Swarm now has direct control over threads and runs through the Agents SDK

​

Architectural Improvements

• Orchestrator Pattern on New Foundation: Agency Swarm’s proven orchestrator-workers pattern now runs on the OpenAI Agents SDK foundation
• Async-first Architecture: Native async execution for better performance
• Full Conversation Persistence: Complete conversation history management (not just thread IDs) for robust session handling
• Enhanced Communication: Agents communicate through defined communication_flows pathways for coordinated multi-agent execution

​

Enhanced Developer Experience

• Structured Outputs: Native Pydantic model support for agent outputs via output_type
• Modern Tool System: @function_tool decorator replaces BaseTool classes for cleaner tool definitions
• Better Validation: output_guardrails and input_guardrails replace the old response_validator system
• Real-time Streaming: Improved streaming capabilities with async response handling

​

Why These Changes? (Architectural Context)

​

Execution Core

• v0.x: Used OpenAI Assistants API runs directly
• v1.x: Uses agents.Runner from OpenAI Agents SDK for more control

​

State Management

• v0.x: Relied on Assistant/Thread objects managed by OpenAI
• v1.x: Uses ThreadManager and ConversationThread managed via RunHooks and shared MasterContext

​

Agent Definition

• v0.x: Custom Agent class with Assistants API integration
• v1.x: agency_swarm.Agent extends agents.Agent, incorporating tools, subagent registration, and file handling

​

Conversation History Persistence

• v0.x (Assistants API): Required thread callbacks for production to persist OpenAI Assistant/Thread IDs. OpenAI managed conversation history server-side.
• v1.x (Agents SDK): Required thread callbacks for production to persist complete conversation histories locally. You manage the full conversation state.

• Callback Structure: threads_callbacks dict → separate load_threads_callback and save_threads_callback parameters
• Data Format: Thread IDs only → Complete conversation histories
• Callback Signatures: Unchanged (both use no-parameter callbacks with closure)

# v0.x
def load_threads(chat_id):
    return load_threads_from_db(chat_id)  # Returns thread IDs

def save_threads(new_threads):
    save_threads_to_db(new_threads)  # Saves thread IDs

agency = Agency(
    agents,
    threads_callbacks={
        'load': lambda: load_threads(chat_id),
        'save': lambda new_threads: save_threads(new_threads)
    }
)

# v1.x
def load_threads(chat_id):
    # Load complete conversation history instead of just thread IDs
    return load_conversation_history(chat_id)

def save_threads(new_threads):
    # Save complete conversation histories
    save_conversation_history(new_threads)

agency = Agency(
    entry_agent,
    load_threads_callback=lambda: load_threads(chat_id),
    save_threads_callback=lambda new_threads: save_threads(new_threads)
)

​

Breaking Changes

​

Quick-Migration Checklist

For a concise overview, consult the table below. See the Step-by-Step Migration section for full details.

agency.get_completion("Hi", response_format={"type": "json_schema", ...})

agent = Agent(response_validator=my_check, ...)

result = agency.get_completion_stream("Hello")

​

Step-by-Step Migration

agency = Agency(
    entry_point_agent,  # Positional argument
    communication_flows=[(agent1, agent2)],
    load_threads_callback=load_threads_callback,
    save_threads_callback=save_threads_callback
)

# v0.x
agent = Agent(temperature=0.7, top_p=1.0, max_completion_tokens=1000)

# v1.x (recommended)
agent = Agent(
    model="gpt-4o",
    model_settings=ModelSettings(
        temperature=0.7,
        top_p=1.0,
        max_completion_tokens=1000,
    )
)

# v0.x - This will CRASH in v1.x
completion = agency.get_completion(
    message="...",
    response_format={"type": "json_schema", "json_schema": {...}}
)
# NotImplementedError: response_format parameter is no longer supported

# v1.x - Required migration
class TaskOutput(BaseModel):
    task_name: str
    status: str

agent = Agent(output_type=TaskOutput, ...)
result = await agency.get_response(message="...", recipient_agent="AgentName")

class MyTool(BaseTool):
    arg1: str = Field(..., description="Description")

    def run(self):
        return f"Result: {self.arg1}"

@function_tool
def my_tool(arg1: str) -> str:
    """Tool description."""
    return f"Result: {arg1}"

# Optional: Tool with context access (if you need shared state)
@function_tool
async def my_tool_with_context(ctx: RunContextWrapper[Any], arg1: str) -> str:
    """Tool with context access."""
    # Access: ctx.context.agents, ctx.context.thread_manager
    return f"Result: {arg1}"

result = agency.get_completion("Hello", recipient_agent="Agent1")

class TestAgent(Agent):
    def __init__(self):
        super().__init__(
            name="TestAgent",
            description="..."
        )

    @override
    def response_validator(self, message):
        # User-defined validation function
        if self.check_errors(message):
            raise ValueError("Error processing message")

        # Returns original message if no errors are raised
        return message

​

Parameter Reference Tables

​

Changes to Agency Class

• v0.x: Used agency_chart parameter where standalone agents were entry points and lists defined communication paths
• v1.x: Entry points passed as positional arguments; communication flows via communication_flows parameter

​

Changes to Agent Class

​

Changes to Tools

​

Changes to Communication

​

Structured Outputs

​

Complete Before/After Example

from agency_swarm import Agency, Agent, BaseTool
from pydantic import Field

# Tool Definition
class ProcessingTool(BaseTool):
    """Processes data."""
    input_data: str = Field(..., description="Data to process")

    def run(self):
        return f"Processed: {self.input_data}"

# Agent with validation
class AnalysisAgent(Agent):
    def __init__(self):
        super().__init__(
            name="AnalysisAgent",
            description="Analyzes data",
            instructions="Process and analyze input data.",
            tools=[ProcessingTool],
            temperature=0.7,
            max_completion_tokens=1000
        )

    def response_validator(self, message):
        if "error" in message.lower():
            raise ValueError("Invalid response detected")
        return message

# Persistence functions
def load_threads(chat_id):
    return load_threads_from_db(chat_id)  # Returns thread IDs

def save_threads(new_threads):
    save_threads_to_db(new_threads)  # Saves thread IDs

# Agency Setup
analyst = AnalysisAgent()
reporter = Agent(
    name="Reporter",
    description="Creates reports",
    instructions="Generate reports from analysis.",
    temperature=0.3
)

agency = Agency(
    agency_chart=[
        analyst,  # Entry point
        [analyst, reporter]  # Communication flow
    ],
    threads_callbacks={
        'load': lambda: load_threads(chat_id),
        'save': lambda threads: save_threads(threads)
    },
    shared_instructions="Be helpful and accurate."
)

# Usage
result = agency.get_completion(
    message="Analyze this data: sample_data",
    response_format={"type": "json_schema", "json_schema": {...}},
    recipient_agent="AnalysisAgent"
)
print(result)

​

Backward Compatibility

• agency_chart: Still works but deprecated; migrate to positional arguments and communication_flows
• get_completion(): Maintained as a synchronous wrapper; deprecated in favor of async get_response()
• get_completion_stream(): Not supported - raises NotImplementedError; use async get_response_stream()
• Agent Parameters: Individual model parameters still work but deprecated; use model_settings
• BaseTool: Temporarily retained for gradual migration to @function_tool

​

Resources

​

Available Examples

• two_agent_conversation.py - Multi-agent communication with automatic thread isolation
• streaming.py - Real-time response streaming with proper event handling
• file_handling.py - File processing and vision analysis using OpenAI’s built-in capabilities
• file_search.py - Vector store creation and FileSearch tool usage with automatic indexing
• file_search_persistence.py - Hosted tool output preservation across conversation turns
• multi_agent_workflow.py - Complex multi-agent collaboration with validation
• response_validation.py - Input and output guardrails implementation with tripwire triggers
• custom_persistence.py - Thread isolation and persistence across application restarts
• chat_completion_provider.py - Custom OpenAI Chat Completions model provider usage

​

Additional Resources

• OpenAI Agents SDK Documentation
• Agency Swarm v0.x Documentation (current production docs)
• GitHub Issues - Report v1.x beta issues with the v1.x beta label