Back to .md Directory

MCP Server Migration Guide

Use this guide when you need to:

May 2, 2026
0 downloads
2 views
mcp workflow
View source

MCP Server Migration Guide

When to Use This Guide

Use this guide when you need to:

  • Migrate from deprecated SSE transport to Streamable HTTP
  • Update your server to comply with MCP specification 2025-03-26
  • Refactor atomic API-oriented tools into workflow-oriented tools
  • Handle breaking changes between MCP specification versions

Overview

The MCP specification evolves to improve security, performance, and developer experience. This guide helps you migrate existing MCP servers to current standards, focusing on three major migration scenarios:

  1. Transport Migration: SSE → Streamable HTTP
  2. Specification Updates: Handling breaking changes
  3. Tool Design Refactoring: Atomic operations → Workflow-oriented tools

Migration Scenario 1: SSE to Streamable HTTP Transport

Why Migrate?

Server-Sent Events (SSE) transport was deprecated in the MCP 2025-03-26 specification in favor of Streamable HTTP because:

  • Better bidirectional communication: Streamable HTTP supports full duplex communication
  • Improved error handling: More robust error propagation and recovery
  • Standard HTTP semantics: Works better with existing infrastructure (load balancers, proxies)
  • OAuth 2.1 integration: Designed to work seamlessly with modern authentication

Migration Steps

Step 1: Update Dependencies

Python (Before):

# requirements.txt
mcp-server-sdk==0.1.0  # Old version with SSE

Python (After):

# requirements.txt
mcp>=1.0.0  # Current version with Streamable HTTP
fastmcp>=0.2.0  # If using FastMCP

TypeScript (Before):

{
  "dependencies": {
    "@modelcontextprotocol/sdk": "^0.1.0"
  }
}

TypeScript (After):

{
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0"
  }
}

Step 2: Update Server Initialization

Python with FastMCP (Before - SSE):

from mcp.server import Server
from mcp.server.sse import SseServerTransport

app = Server("my-server")
transport = SseServerTransport("/sse")

# SSE-specific configuration
@app.route("/sse")
async def sse_endpoint(request):
    return transport.handle_sse(request)

Python with FastMCP (After - Streamable HTTP):

from mcp.server.fastmcp import FastMCP

# FastMCP handles Streamable HTTP automatically
mcp = FastMCP("my-server")

# No manual transport configuration needed
# FastMCP uses Streamable HTTP by default

TypeScript (Before - SSE):

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";

const server = new Server(
  {
    name: "my-server",
    version: "1.0.0",
  },
  {
    capabilities: {},
  }
);

const transport = new SSEServerTransport("/messages", server);

TypeScript (After - Streamable HTTP):

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamable-http.js";

const server = new Server(
  {
    name: "my-server",
    version: "1.0.0",
  },
  {
    capabilities: {},
  }
);

const transport = new StreamableHTTPServerTransport({
  server,
  endpoint: "/mcp/v1",
});

Step 3: Update HTTP Endpoints

Before (SSE):

  • SSE endpoint: GET /sse (for event stream)
  • Message endpoint: POST /messages (for client messages)

After (Streamable HTTP):

  • Single endpoint: POST /mcp/v1 (handles all communication)
  • Optional: GET /.well-known/oauth-protected-resource (for OAuth metadata)

Express.js Example (Before - SSE):

app.get("/sse", (req, res) => {
  transport.handleSSE(req, res);
});

app.post("/messages", async (req, res) => {
  await transport.handleMessage(req, res);
});

Express.js Example (After - Streamable HTTP):

app.post("/mcp/v1", async (req, res) => {
  await transport.handle(req, res);
});

// Optional: OAuth metadata endpoint
app.get("/.well-known/oauth-protected-resource", (req, res) => {
  res.json({
    resource: "https://your-server.com",
    authorization_servers: ["https://auth.your-server.com"],
    scopes_supported: ["read:data", "write:data"],
  });
});

Step 4: Update Client Configuration

Claude Desktop Config (Before - SSE):

{
  "mcpServers": {
    "my-server": {
      "url": "https://api.example.com/sse",
      "transport": "sse"
    }
  }
}

Claude Desktop Config (After - Streamable HTTP):

{
  "mcpServers": {
    "my-server": {
      "url": "https://api.example.com/mcp/v1",
      "transport": "streamable-http"
    }
  }
}

Step 5: Update Error Handling

SSE and Streamable HTTP handle errors differently.

Before (SSE):

# SSE sent errors as events
async def handle_error(error):
    return {
        "event": "error",
        "data": json.dumps({"message": str(error)})
    }

After (Streamable HTTP):

# Streamable HTTP uses standard JSON-RPC error responses
from mcp.types import ErrorData, McpError

async def handle_error(error):
    raise McpError(
        code=-32603,  # Internal error
        message="Operation failed",
        data=ErrorData(
            details=str(error),
            help_url="https://docs.example.com/errors"
        )
    )

Step 6: Test the Migration

# Test with MCP Inspector
npx @modelcontextprotocol/inspector https://api.example.com/mcp/v1

# Test with curl
curl -X POST https://api.example.com/mcp/v1 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {}
  }'

Common Migration Issues

Issue 1: Connection Timeouts

  • Cause: SSE kept connections open indefinitely; Streamable HTTP uses request-response
  • Solution: Implement proper connection pooling and retry logic on the client side

Issue 2: Event Streaming

  • Cause: SSE pushed events to clients; Streamable HTTP is request-driven
  • Solution: Use polling or webhooks for server-initiated notifications

Issue 3: CORS Configuration

  • Cause: Different CORS requirements for POST vs GET+POST
  • Solution: Update CORS to allow POST to /mcp/v1 endpoint
# Python with FastAPI
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://claude.ai"],
    allow_methods=["POST", "OPTIONS"],
    allow_headers=["Content-Type", "Authorization"],
)

Migration Scenario 2: Specification Breaking Changes

MCP 2025-03-26 Breaking Changes

Change 1: Capability Negotiation

Before (Pre-2025-03-26):

// Capabilities were optional
const server = new Server({
  name: "my-server",
  version: "1.0.0",
});

After (2025-03-26):

// Capabilities must be explicitly declared
const server = new Server(
  {
    name: "my-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
      resources: {},
      prompts: {},
    },
  }
);

Migration Action: Add explicit capability declarations to your server initialization.

Change 2: Tool Input Schema Validation

Before:

# Input validation was optional
@mcp.tool()
async def create_task(title: str, description: str):
    return {"id": 123, "title": title}

After:

# Input schema must be explicitly defined
from pydantic import BaseModel, Field

class CreateTaskInput(BaseModel):
    title: str = Field(..., description="Task title")
    description: str = Field(..., description="Task description")

@mcp.tool()
async def create_task(input: CreateTaskInput) -> dict:
    return {"id": 123, "title": input.title}

Migration Action: Add Pydantic models (Python) or Zod schemas (TypeScript) for all tool inputs.

Change 3: Error Code Standardization

Before:

# Custom error codes
raise Exception("Something went wrong")  # Became generic -32603

After:

# Use standard JSON-RPC error codes
from mcp.types import McpError

# -32600: Invalid Request
# -32601: Method not found
# -32602: Invalid params
# -32603: Internal error
# -32700: Parse error

raise McpError(
    code=-32602,
    message="Invalid parameters",
    data={"field": "title", "issue": "required"}
)

Migration Action: Replace generic exceptions with specific JSON-RPC error codes.

Change 4: OAuth 2.1 Requirement

Before:

# OAuth was optional for remote servers
@app.post("/mcp/v1")
async def handle_request(request):
    # No authentication
    return await process_request(request)

After:

# OAuth 2.1 required for remote servers with user data
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def verify_token(credentials = Depends(security)):
    token = credentials.credentials
    # Validate JWT token
    if not is_valid_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

@app.post("/mcp/v1")
async def handle_request(request, token = Depends(verify_token)):
    return await process_request(request, token)

Migration Action: Implement OAuth 2.1 authentication for remote servers handling user-specific data.

Step-by-Step Specification Update Procedure

  1. Review Changelog: Read the MCP specification changelog for your version
  2. Update Dependencies: Install latest SDK versions
  3. Run Deprecation Checks: Look for deprecation warnings in logs
  4. Update Code: Apply breaking changes systematically
  5. Update Tests: Ensure tests cover new requirements
  6. Test with Inspector: Validate against new specification
  7. Deploy Gradually: Use canary deployments to test with real traffic

Migration Scenario 3: Tool Design Refactoring

From Atomic to Workflow-Oriented Tools

Anti-Pattern: Atomic API Operations

Before (❌ Atomic Tools):

@mcp.tool()
async def github_get_user(username: str):
    """Get a GitHub user"""
    return await github_api.get(f"/users/{username}")

@mcp.tool()
async def github_get_repos(username: str):
    """Get user's repositories"""
    return await github_api.get(f"/users/{username}/repos")

@mcp.tool()
async def github_get_issues(repo: str):
    """Get repository issues"""
    return await github_api.get(f"/repos/{repo}/issues")

@mcp.tool()
async def github_create_issue(repo: str, title: str, body: str):
    """Create an issue"""
    return await github_api.post(f"/repos/{repo}/issues", {
        "title": title,
        "body": body
    })

Problems:

  • LLM must make multiple tool calls to accomplish a task
  • Requires LLM to understand API relationships
  • Inefficient (multiple round trips)
  • Error-prone (LLM might call tools in wrong order)

Best Practice: Workflow-Oriented Tools

After (✅ Workflow Tools):

from pydantic import BaseModel, Field
from typing import Optional, List

class SearchGitHubInput(BaseModel):
    query: str = Field(..., description="Search query (e.g., 'user:octocat stars:>100')")
    type: str = Field("repositories", description="Search type: repositories, issues, users")
    limit: int = Field(10, description="Maximum results to return")

@mcp.tool()
async def search_github(input: SearchGitHubInput) -> dict:
    """
    Search GitHub for repositories, issues, or users.

    This tool handles the complete search workflow including:
    - Query parsing and validation
    - API pagination
    - Result formatting
    - Error handling

    Examples:
    - Find popular Python repos: "language:python stars:>1000"
    - Find user's repos: "user:octocat"
    - Find open issues: "is:issue is:open label:bug"
    """
    results = await github_api.search(
        type=input.type,
        query=input.query,
        per_page=input.limit
    )
    return {
        "total_count": results["total_count"],
        "items": results["items"][:input.limit]
    }

class CreateIssueInput(BaseModel):
    repository: str = Field(..., description="Repository in format 'owner/repo'")
    title: str = Field(..., description="Issue title")
    body: str = Field(..., description="Issue description")
    labels: Optional[List[str]] = Field(None, description="Labels to apply")
    assignees: Optional[List[str]] = Field(None, description="Users to assign")

@mcp.tool()
async def create_github_issue(input: CreateIssueInput) -> dict:
    """
    Create a GitHub issue with full workflow support.

    This tool handles the complete issue creation workflow:
    - Repository validation
    - Label and assignee validation
    - Issue creation
    - Automatic linking to related issues
    - Notification handling

    Returns the created issue with URL and number.
    """
    # Validate repository exists
    repo = await github_api.get(f"/repos/{input.repository}")

    # Create issue with all metadata
    issue = await github_api.post(f"/repos/{input.repository}/issues", {
        "title": input.title,
        "body": input.body,
        "labels": input.labels or [],
        "assignees": input.assignees or []
    })

    return {
        "number": issue["number"],
        "url": issue["html_url"],
        "state": issue["state"]
    }

class AnalyzeRepositoryInput(BaseModel):
    repository: str = Field(..., description="Repository in format 'owner/repo'")
    include_issues: bool = Field(True, description="Include issue analysis")
    include_contributors: bool = Field(True, description="Include contributor stats")

@mcp.tool()
async def analyze_github_repository(input: AnalyzeRepositoryInput) -> dict:
    """
    Analyze a GitHub repository comprehensively.

    This tool provides a complete repository analysis including:
    - Basic repository information (stars, forks, language)
    - Recent activity and commit frequency
    - Open issues and pull requests (if requested)
    - Top contributors (if requested)
    - Health indicators (has README, license, etc.)

    Perfect for understanding a repository at a glance.
    """
    repo = await github_api.get(f"/repos/{input.repository}")

    analysis = {
        "name": repo["full_name"],
        "description": repo["description"],
        "stars": repo["stargazers_count"],
        "forks": repo["forks_count"],
        "language": repo["language"],
        "health": {
            "has_readme": repo.get("has_readme", False),
            "has_license": repo["license"] is not None,
            "has_wiki": repo["has_wiki"]
        }
    }

    if input.include_issues:
        issues = await github_api.get(f"/repos/{input.repository}/issues")
        analysis["open_issues"] = len([i for i in issues if i["state"] == "open"])

    if input.include_contributors:
        contributors = await github_api.get(f"/repos/{input.repository}/contributors")
        analysis["top_contributors"] = [
            {"login": c["login"], "contributions": c["contributions"]}
            for c in contributors[:5]
        ]

    return analysis

Benefits:

  • Single tool call accomplishes user intent
  • Better error handling and validation
  • More efficient (fewer round trips)
  • Clearer tool descriptions help LLM choose correctly
  • Easier to maintain and test

Refactoring Procedure

Step 1: Identify User Workflows

Map your atomic tools to actual user intents:

Atomic Tools:
- get_user
- get_repos
- get_issues
- create_issue

User Workflows:
- "Find repositories by a user" → search_github
- "Create an issue in a repo" → create_github_issue
- "Analyze a repository" → analyze_github_repository

Step 2: Group Related Operations

# Group 1: Search and Discovery
- search_github (replaces: get_user, get_repos, search_repos)

# Group 2: Issue Management
- create_github_issue (replaces: create_issue, add_labels, assign_users)
- update_github_issue (replaces: update_issue, close_issue, reopen_issue)

# Group 3: Repository Analysis
- analyze_github_repository (replaces: get_repo, get_stats, get_contributors)

Step 3: Design Comprehensive Input Schemas

# Bad: Minimal schema
class CreateIssueInput(BaseModel):
    repo: str
    title: str

# Good: Comprehensive schema
class CreateIssueInput(BaseModel):
    repository: str = Field(
        ...,
        description="Repository in format 'owner/repo'",
        pattern=r"^[\w-]+/[\w-]+$"
    )
    title: str = Field(
        ...,
        description="Issue title (max 256 characters)",
        max_length=256
    )
    body: str = Field(
        ...,
        description="Issue description (supports Markdown)"
    )
    labels: Optional[List[str]] = Field(
        None,
        description="Labels to apply (e.g., ['bug', 'urgent'])"
    )
    assignees: Optional[List[str]] = Field(
        None,
        description="GitHub usernames to assign"
    )
    milestone: Optional[int] = Field(
        None,
        description="Milestone number to associate"
    )

Step 4: Write Detailed Tool Descriptions

# Bad: Minimal description
@mcp.tool()
async def create_issue(input: CreateIssueInput):
    """Create an issue"""
    pass

# Good: Comprehensive description
@mcp.tool()
async def create_github_issue(input: CreateIssueInput):
    """
    Create a GitHub issue with full workflow support.

    This tool handles the complete issue creation workflow including:
    - Repository validation (checks if repo exists and is accessible)
    - Label validation (verifies labels exist in the repository)
    - Assignee validation (checks if users can be assigned)
    - Issue creation with all metadata
    - Automatic linking to related issues (if mentioned in body)
    - Notification handling

    Use this tool when the user wants to:
    - Report a bug in a repository
    - Request a new feature
    - Ask a question to repository maintainers
    - Track a task or todo item

    Examples:
    - Create a bug report: repository="owner/repo", title="Bug: App crashes on startup", labels=["bug"]
    - Request a feature: repository="owner/repo", title="Feature: Add dark mode", labels=["enhancement"]

    Returns:
    - Issue number (for reference)
    - Issue URL (for sharing)
    - Issue state (usually "open")

    Errors:
    - 404: Repository not found or not accessible
    - 422: Invalid labels or assignees
    - 403: No permission to create issues
    """
    pass

Step 5: Implement and Test

# Test workflow tools with realistic scenarios
async def test_create_issue_workflow():
    # Test complete workflow
    result = await create_github_issue(CreateIssueInput(
        repository="octocat/Hello-World",
        title="Test issue",
        body="This is a test",
        labels=["bug"],
        assignees=["octocat"]
    ))

    assert result["number"] > 0
    assert "url" in result
    assert result["state"] == "open"

# Test with MCP Inspector
# The LLM should be able to create an issue in one tool call

Step 6: Deprecate Atomic Tools Gradually

@mcp.tool()
async def github_get_user(username: str):
    """
    [DEPRECATED] Get a GitHub user.

    This tool is deprecated. Use search_github with type="users" instead.

    Migration example:
    Old: github_get_user(username="octocat")
    New: search_github(query="user:octocat", type="users")
    """
    # Keep for backward compatibility during migration
    return await github_api.get(f"/users/{username}")

Migration Checklist

  • Identify all atomic tools in your server
  • Map atomic tools to user workflows
  • Design workflow-oriented tools with comprehensive schemas
  • Write detailed tool descriptions with examples
  • Implement workflow tools with proper error handling
  • Test with MCP Inspector (verify single tool call accomplishes task)
  • Update documentation and examples
  • Deprecate atomic tools with migration guidance
  • Monitor usage and remove deprecated tools after transition period

Testing Migrated Servers

Test Plan Template

# Migration Test Plan

## Pre-Migration Tests

- [ ] Document all existing tool calls and expected outputs
- [ ] Create test suite for current functionality
- [ ] Measure baseline performance (latency, error rate)

## Migration Tests

- [ ] Verify all tools are accessible via new transport
- [ ] Test capability negotiation
- [ ] Validate input schemas with valid and invalid data
- [ ] Test error handling and error codes
- [ ] Verify OAuth token validation (if applicable)
- [ ] Test with MCP Inspector

## Post-Migration Tests

- [ ] Verify all workflows still function correctly
- [ ] Compare performance metrics (should be similar or better)
- [ ] Test with real LLM clients (Claude Desktop, etc.)
- [ ] Monitor error logs for unexpected issues
- [ ] Validate backward compatibility (if maintaining old endpoints)

## Rollback Plan

- [ ] Document rollback procedure
- [ ] Keep old version running in parallel during transition
- [ ] Set up monitoring and alerts
- [ ] Define success criteria for full cutover

Testing with MCP Inspector

# Test Streamable HTTP server
npx @modelcontextprotocol/inspector https://api.example.com/mcp/v1

# Test with OAuth
npx @modelcontextprotocol/inspector \
  https://api.example.com/mcp/v1 \
  --auth-token "your-access-token"

# Test specific tool
# In Inspector UI:
# 1. Go to Tools tab
# 2. Select your migrated tool
# 3. Fill in test inputs
# 4. Click "Execute"
# 5. Verify output matches expected format

Automated Testing

# Python test example
import pytest
from mcp.client import Client

@pytest.mark.asyncio
async def test_migrated_tool():
    client = Client("https://api.example.com/mcp/v1")

    # Test tool is available
    tools = await client.list_tools()
    assert "create_github_issue" in [t.name for t in tools]

    # Test tool execution
    result = await client.call_tool("create_github_issue", {
        "repository": "test/repo",
        "title": "Test issue",
        "body": "Test body"
    })

    assert result["number"] > 0
    assert "url" in result

Common Migration Pitfalls

Pitfall 1: Incomplete Error Migration

Problem: Forgetting to update error handling for new transport.

Solution: Use standard JSON-RPC error codes consistently.

# ❌ Bad: Generic exceptions
raise Exception("Something went wrong")

# ✅ Good: Specific JSON-RPC errors
from mcp.types import McpError

raise McpError(
    code=-32602,
    message="Invalid repository format",
    data={"expected": "owner/repo", "received": repo}
)

Pitfall 2: Breaking Client Compatibility

Problem: Changing tool names or schemas without versioning.

Solution: Use versioning or maintain backward compatibility.

# Option 1: Version in tool name
@mcp.tool()
async def create_issue_v2(input: CreateIssueInputV2):
    """New version with enhanced features"""
    pass

# Option 2: Maintain both during transition
@mcp.tool()
async def create_issue(input: CreateIssueInput):
    """[DEPRECATED] Use create_github_issue instead"""
    return await create_github_issue(input)

@mcp.tool()
async def create_github_issue(input: CreateIssueInput):
    """Current version"""
    pass

Pitfall 3: Insufficient Testing

Problem: Not testing edge cases and error conditions.

Solution: Create comprehensive test suite covering all scenarios.

# Test matrix
test_cases = [
    # Happy path
    {"repo": "owner/repo", "title": "Test", "expected": "success"},
    # Invalid repo format
    {"repo": "invalid", "title": "Test", "expected": "error"},
    # Missing required field
    {"repo": "owner/repo", "expected": "error"},
    # Empty title
    {"repo": "owner/repo", "title": "", "expected": "error"},
    # Very long title
    {"repo": "owner/repo", "title": "x" * 1000, "expected": "error"},
]

Pitfall 4: Ignoring Performance Impact

Problem: New transport or tool design causes performance regression.

Solution: Benchmark before and after migration.

import time

async def benchmark_tool(tool_name, input_data, iterations=100):
    start = time.time()
    for _ in range(iterations):
        await client.call_tool(tool_name, input_data)
    end = time.time()

    avg_latency = (end - start) / iterations
    print(f"{tool_name}: {avg_latency:.3f}s average latency")

Migration Timeline Template

Week 1: Planning and Preparation

  • Review MCP specification changes
  • Audit existing tools and identify migration needs
  • Create migration plan and test strategy
  • Set up staging environment

Week 2: Implementation

  • Update dependencies
  • Migrate transport layer (SSE → Streamable HTTP)
  • Update tool schemas and error handling
  • Implement OAuth if required

Week 3: Testing

  • Run automated test suite
  • Test with MCP Inspector
  • Perform integration testing with real clients
  • Load testing and performance validation

Week 4: Deployment

  • Deploy to staging environment
  • Canary deployment (10% traffic)
  • Monitor metrics and error rates
  • Gradual rollout (25% → 50% → 100%)
  • Deprecate old endpoints

Week 5: Cleanup

  • Remove deprecated code
  • Update documentation
  • Announce migration completion
  • Post-mortem and lessons learned

Getting Help

If you encounter issues during migration:

  1. Check MCP Specification: https://spec.modelcontextprotocol.io/
  2. Review SDK Documentation:
  3. Use MCP Inspector: Debug issues interactively
  4. Community Support: Join MCP Discord or GitHub Discussions
  5. Load Relevant Steering Files:
    • transport-protocols.md for transport issues
    • oauth-authentication.md for auth issues
    • error-handling.md for error-related problems
    • testing-with-inspector.md for testing help

Summary

Successful migration requires:

  • ✅ Systematic approach following documented procedures
  • ✅ Comprehensive testing at each stage
  • ✅ Gradual rollout with monitoring
  • ✅ Clear communication with users
  • ✅ Rollback plan in case of issues

The effort invested in proper migration pays off through:

  • Better security (OAuth 2.1, HTTPS)
  • Improved reliability (Streamable HTTP)
  • Enhanced developer experience (workflow-oriented tools)
  • Future-proof architecture (specification compliance)

Related Documents