MCP Server Setup Skill

This skill provides guidance for setting up a stateless MCP server using the Official MCP SDK in FastAPI.

Purpose

Create a stateless MCP server that provides 5 task management tools (add_task, list_tasks, complete_task, delete_task, update_task) with proper user isolation, database integration, and error handling.

When to Use

Use this skill when:

• Setting up a new MCP server for chatbot integration
• Adding MCP tools for task management functionality
• Creating MCP server infrastructure with database integration
• Implementing tool chaining and error handling for MCP

Capabilities

• Stateless Design: Each request independent, no server-side state between calls
• User Isolation: All tools require user_id and enforce ownership at query level
• Tool Chaining: Tools designed to be safely called by other tools
• Async Performance: Full async/await support for concurrent operations
• Error Handling: Structured errors (400, 401, 403, 404, 500) with descriptive messages

MCP Server Architecture

Server Structure

backend/
├── mcp_server.py      # Main MCP server with FastAPI integration
├── mcp_tools.py       # Tool implementations
├── mcp_config.py      # Server configuration
└── main.py            # Updated with MCP server startup

Tool Signatures

Implementation Pattern

MCP Server Setup

from mcp.server.fastapi import FastAPIMCP
from fastapi import FastAPI

app = FastAPI()
mcp_server = FastAPIMCP(app)

@mcp_server.tool()
async def add_task(user_id: str, title: str, description: str = None) -> dict:
    """Add a new task for the specified user."""
    # Extract user_id from context
    # Validate parameters
    # Execute with user_id filter: WHERE user_id = :user_id
    # Return structured JSON response

Database Session Management

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(DATABASE_URL)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db_session():
    async with async_session() as session:
        yield session

User ID Enforcement

async def add_task(user_id: str, title: str, ...) -> dict:
    if not user_id:
        raise ValueError("user_id is required")

    # Always filter by user_id in queries
    statement = select(Task).where(
        Task.id == task_id,
        Task.user_id == user_id  # Critical for isolation
    )

Tool Implementation Checklist

For each tool, implement:

1. Parameter Validation: Check required fields, types, and constraints
2. User ID Extraction: Get user_id from request context or parameters
3. Database Operation: Execute with proper user_id filtering
4. Response Construction: Return structured JSON with success status
5. Error Handling: Catch exceptions and return appropriate error codes
6. Logging: Log operation outcomes for debugging

Error Taxonomy

Tool Chaining Support

Design tools to support chaining:

async def complete_task(user_id: str, task_id: str) -> dict:
    """Mark a task as complete. Can be called by other tools."""
    task = await get_task_by_id(user_id, task_id)
    if task:
        task.completed = True
        await session.commit()
    return {"success": True, "task": task_to_dict(task)}

async def list_tasks(user_id: str, status: str = None) -> dict:
    """List tasks. Supports filtering by status."""
    # This can chain to complete_task or other tools

Dependencies

mcp>=0.9.0
fastapi>=0.100.0
sqlmodel>=0.0.14
sqlalchemy>=2.0.0

Running the Server

cd backend
uvicorn mcp_server:app --host 0.0.0.0 --port 8001

Integration with Main App

# main.py
from mcp_server import mcp

app = FastAPI()
app.include_router(mcp.router)

@app.on_event("startup")
async def startup():
    await mcp.run()

Verification Checklist

• All 5 tools registered and accessible
• User ID enforced in every database query
• JSON responses match specification
• Error handling returns appropriate codes
• Tool chaining works correctly
• Server starts without errors
• Database connections work