Viết Custom MCP Server

pip install mcp httpx

# weather_server.py
from mcp.server.fastmcp import FastMCP
import httpx
 
mcp = FastMCP("Weather Service")
 
@mcp.tool()
async def get_weather(city: str) -> str:
    """
    Get current weather for a city.
    
    Args:
        city: City name (e.g., "Hanoi", "Tokyo", "New York")
    
    Returns:
        Current weather information
    """
    async with httpx.AsyncClient() as client:
        # Dùng wttr.in API (free, no key needed)
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        
        if response.status_code == 200:
            data = response.json()
            current = data["current_condition"][0]
            
            return f"""
Weather in {city}:
- Temperature: {current['temp_C']}°C ({current['temp_F']}°F)
- Condition: {current['weatherDesc'][0]['value']}
- Humidity: {current['humidity']}%
- Wind: {current['windspeedKmph']} km/h
"""
        else:
            return f"Could not fetch weather for {city}"
 
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
    """
    Get weather forecast for upcoming days.
    
    Args:
        city: City name
        days: Number of days (1-5)
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        
        if response.status_code != 200:
            return f"Could not fetch forecast for {city}"
        
        data = response.json()
        forecast_data = data.get("weather", [])[:days]
        
        result = [f"Forecast for {city}:"]
        for day in forecast_data:
            date = day["date"]
            max_temp = day["maxtempC"]
            min_temp = day["mintempC"]
            desc = day["hourly"][4]["weatherDesc"][0]["value"]
            result.append(f"- {date}: {min_temp}°C - {max_temp}°C, {desc}")
        
        return "\n".join(result)
 
@mcp.resource("weather://cities")
def list_popular_cities() -> list:
    """List popular cities for weather queries."""
    return [
        "Hanoi", "Ho Chi Minh City", "Da Nang",
        "Tokyo", "Seoul", "Singapore",
        "New York", "London", "Paris"
    ]
 
if __name__ == "__main__":
    mcp.run()

# Chạy server
python weather_server.py
 
# Hoặc dùng MCP Inspector
npx @modelcontextprotocol/inspector python weather_server.py

{
  "mcpServers": {
    "weather": {
      "command": "python3",
      "args": ["/absolute/path/to/weather_server.py"]
    }
  }
}

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
 
mcp = FastMCP("SQLite Query Server")
 
DB_PATH = "/path/to/database.db"
 
@mcp.tool()
def query_database(sql: str) -> str:
    """
    Execute a SELECT query on the database.
    Only SELECT queries are allowed for safety.
    """
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are allowed"
    
    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        
        # Format as table
        result = [" | ".join(columns)]
        result.append("-" * len(result[0]))
        for row in rows:
            result.append(" | ".join(str(cell) for cell in row))
        
        return "\n".join(result)
    except Exception as e:
        return f"Error: {str(e)}"
    finally:
        conn.close()
 
@mcp.resource("schema://tables")
def list_tables() -> str:
    """List all tables in the database."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    )
    tables = [row[0] for row in cursor.fetchall()]
    conn.close()
    return "\n".join(tables)
 
if __name__ == "__main__":
    mcp.run()

# api_server.py
from mcp.server.fastmcp import FastMCP
import httpx
import os
 
mcp = FastMCP("Authenticated API Server")
 
# Lấy API key từ environment
API_KEY = os.environ.get("MY_API_KEY", "")
 
@mcp.tool()
async def call_protected_api(endpoint: str) -> str:
    """Call a protected API endpoint."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.example.com/{endpoint}",
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        return response.json()
 
if __name__ == "__main__":
    mcp.run()

{
  "my-api": {
    "command": "python3",
    "args": ["/path/to/api_server.py"],
    "env": {
      "MY_API_KEY": "sk-xxxxx"
    }
  }
}

// server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
 
const server = new Server(
  { name: "my-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);
 
server.setRequestHandler("tools/list", async () => ({
  tools: [{
    name: "greet",
    description: "Say hello to someone",
    inputSchema: {
      type: "object",
      properties: {
        name: { type: "string", description: "Name to greet" }
      },
      required: ["name"]
    }
  }]
}));
 
server.setRequestHandler("tools/call", async (request) => {
  if (request.params.name === "greet") {
    const name = request.params.arguments.name;
    return { content: [{ type: "text", text: `Hello, ${name}!` }] };
  }
  throw new Error("Unknown tool");
});
 
const transport = new StdioServerTransport();
await server.connect(transport);

@mcp.tool()
def risky_operation() -> str:
    try:
        # Do something
        return "Success"
    except Exception as e:
        return f"Error: {str(e)}"  # Không raise, return error message

import asyncio
 
@mcp.tool()
async def slow_operation() -> str:
    try:
        result = await asyncio.wait_for(
            do_something(),
            timeout=30.0
        )
        return result
    except asyncio.TimeoutError:
        return "Operation timed out"

@mcp.tool()
def safe_file_read(path: str) -> str:
    # Validate path
    if ".." in path or path.startswith("/"):
        return "Error: Invalid path"
    
    allowed_dir = "/safe/directory"
    full_path = os.path.join(allowed_dir, path)
    
    if not full_path.startswith(allowed_dir):
        return "Error: Path traversal detected"
    
    return open(full_path).read()

from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("My Custom Server")
 
# TODO: Add your tools here
@mcp.tool()
def my_tool(param: str) -> str:
    """Description of what this tool does."""
    return "Result"
 
# TODO: Add your resources here
@mcp.resource("my://data")
def my_resource() -> dict:
    """Description of this resource."""
    return {"key": "value"}
 
if __name__ == "__main__":
    mcp.run()