Loading...
Loading...
Loading...
# 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)**:
```python
# requirements.txt
mcp-server-sdk==0.1.0 # Old version with SSE
```
**Python (After)**:
```python
# requirements.txt
mcp>=1.0.0 # Current version with Streamable HTTP
fastmcp>=0.2.0 # If using FastMCP
```
**TypeScript (Before)**:
```json
{
"dependencies": {
"@modelcontextprotocol/sdk": "^0.1.0"
}
}
```
**TypeScript (After)**:
```json
{
"dependencies": {
"@modelcontextprotocol/sdk": "^1.0.0"
}
}
```
#### Step 2: Update Server Initialization
**Python with FastMCP (Before - SSE)**:
```python
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)**:
```python
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)**:
```typescript
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)**:
```typescript
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)**:
```typescript
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)**:
```typescript
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)**:
```json
{
"mcpServers": {
"my-server": {
"url": "https://api.example.com/sse",
"transport": "sse"
}
}
}
```
**Claude Desktop Config (After - Streamable HTTP)**:
```json
{
"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)**:
```python
# SSE sent errors as events
async def handle_error(error):
return {
"event": "error",
"data": json.dumps({"message": str(error)})
}
```
**After (Streamable HTTP)**:
```python
# 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
```bash
# 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
# 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)**:
```typescript
// Capabilities were optional
const server = new Server({
name: "my-server",
version: "1.0.0",
});
```
**After (2025-03-26)**:
```typescript
// 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**:
```python
# Input validation was optional
@mcp.tool()
async def create_task(title: str, description: str):
return {"id": 123, "title": title}
```
**After**:
```python
# 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**:
```python
# Custom error codes
raise Exception("Something went wrong") # Became generic -32603
```
**After**:
```python
# 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**:
```python
# OAuth was optional for remote servers
@app.post("/mcp/v1")
async def handle_request(request):
# No authentication
return await process_request(request)
```
**After**:
```python
# 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)**:
```python
@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)**:
```python
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
```python
# 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
```python
# 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
```python
# 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
```python
# 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
```python
@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
```markdown
# 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
```bash
# 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
# 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.
```python
# ❌ 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.
```python
# 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.
```python
# 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.
```python
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**:
- Python: https://github.com/modelcontextprotocol/python-sdk
- TypeScript: https://github.com/modelcontextprotocol/typescript-sdk
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)
어떠한 문서나 스크립트가 다른 **프로토콜 / 포트 / 호스트** 에 있는 리소스 사용하는 것을 제한하는 정책. 예를 들어, 다음과 같은 사이트에서 리소스를 다른 곳으로 요청한다고 하자.
* **Production MDB**: updated monthly.
This document outlines the mandatory procedures for developing and verifying VCR elements (shaders, manifests, and assets) to ensure high-fidelity, centered, and non-clipping renders.
http://localhost:8000