Building MCP Tooling

pip install fastmcp uvicorn

from fastmcp import FastMCP

mcp = FastMCP("My MCP Server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Start with: uvicorn main:mcp --port 8000

from fastmcp import FastMCP
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# Create FastAPI app
app = FastAPI()

# Add CORS middleware for browser clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# Create MCP server
mcp = FastMCP("My MCP Server")

@mcp.tool()
async def multiply(a: float, b: float) -> float:
    """Multiply two numbers"""
    return a * b

# Mount MCP as ASGI app
mcp_app = mcp.http_app(path='/mcp')
app.mount("/", mcp_app)

# Add health check
@app.get("/health")
async def health():
    return {"status": "healthy"}

# Start: uvicorn main:app --reload
# MCP URL: http://localhost:8000/mcp

from fastmcp import FastMCP
from fastmcp.server.dependencies import get_http_request
from fastapi import HTTPException, status

mcp = FastMCP("Protected Server")

# API key management
API_KEYS = {"secret-key-1": "Production API Key"}

@mcp.tool()
async def protected_function(data: str) -> dict:
    """Protected function requiring authentication"""
    request = get_http_request()
    
    # Check API key from header
    api_key = request.headers.get("X-API-Key")
    
    if api_key not in API_KEYS:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key"
        )
    
    return {"status": "success", "data": data}

@mcp.resource("config://settings")
async def get_settings():
    """Get current application settings"""
    return {
        "version": "1.0.0",
        "environment": "production",
        "features": ["api", "auth", "logging"]
    }

@mcp.resource("data://{category}/{id}")
async def get_data(category: str, id: str):
    """Get data by category and ID"""
    # Fetch from database
    data = await fetch_from_db(category, id)
    return data

import httpx
from typing import Optional, Dict, Any

@mcp.tool()
async def api_call(
    endpoint: str, 
    method: str = "GET",
    data: Optional[Dict[str, Any]] = None
) -> dict:
    """Make HTTP request with error handling"""
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            response = await client.request(
                method=method,
                url=endpoint,
                json=data
            )
            response.raise_for_status()
            
            return {
                "success": True,
                "data": response.json(),
                "status_code": response.status_code
            }
            
        except httpx.HTTPStatusError as e:
            return {
                "success": False,
                "error": f"HTTP {e.response.status_code}",
                "status_code": e.response.status_code
            }
        except httpx.TimeoutException:
            return {
                "success": False,
                "error": "Request timeout"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

from fastmcp import Context

@mcp.tool()
async def process_data(file_path: str, ctx: Context) -> dict:
    """Process file with progress reporting"""
    # Log information
    await ctx.info(f"Processing file: {file_path}")
    
    # Read file
    data = await read_file(file_path)
    
    # Report progress
    await ctx.report_progress(50, f"Processing {len(data)} items")
    
    # Process data
    results = await process_items(data)
    
    await ctx.report_progress(100, "Processing complete")
    
    return {"processed": len(results), "results": results}

@mcp.tool()
async def send_email(to: str, subject: str, body: str) -> dict:
    """
    Send an email notification.
    
    Use this when:
    - User explicitly requests to send an email
    - System needs to send notifications
    - Alerts need to be triggered
    
    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email content (plain text)
    
    Returns:
        Dict with status and message_id
    """
    # Implementation

from pydantic import BaseModel
from datetime import datetime

class APIResponse(BaseModel):
    success: bool
    data: Any = None
    error: str = None
    timestamp: datetime
    
@mcp.tool()
async def fetch_data(query: str) -> APIResponse:
    """Fetch data with structured response"""
    try:
        result = await database.query(query)
        return APIResponse(
            success=True,
            data=result,
            timestamp=datetime.now()
        )
    except Exception as e:
        return APIResponse(
            success=False,
            error=str(e),
            timestamp=datetime.now()
        )

from fastmcp.server.dependencies import get_http_request
from typing import Annotated
from fastapi import Header

@mcp.tool()
async def database_query(
    query: str,
    db_url: Annotated[str, Header(alias="X-Database-URL")]
) -> list:
    """Execute database query with injected connection"""
    # Use db_url from header
    conn = await get_connection(db_url)
    return await conn.fetch(query)

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]