Building Custom Server

Khái niệm cơ bản

Khi không có server có sẵn cho use case của bạn, hãy viết server riêng! MCP SDK hỗ trợ Python và TypeScript.

---

Quick Start: Weather API Server

Mục tiêu

Viết server cung cấp thông tin thời tiết realtime cho Claude.

Bước 1: Cài đặt dependencies

pip install mcp httpx

Bước 2: Viết server code

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()

Bước 3: Test server locally

# Chạy server
python weather_server.py

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

Bước 4: Config Claude Desktop

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

Bước 5: Test với Claude

• “Thời tiết Hà Nội hôm nay thế nào?”
• “Dự báo 3 ngày tới ở Tokyo”
• “So sánh thời tiết Hà Nội và Sài Gòn”

---

Server với Database

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()

---

Server với API Authentication

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()

Config với env:

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

---

TypeScript Server

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);

---

Best Practices

1. Error Handling

@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

2. Timeout

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"

3. Input Validation

@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()

---

Bài tập thực hành

Mục tiêu

Viết MCP Server cho một use case thực tế của bạn.

Gợi ý ideas

1. Crypto Price Server

   • Tool: get_price(symbol: str) → Giá BTC, ETH…
   • Resource: crypto://trending → Top 10 coins

2. News Server

   • Tool: search_news(topic: str) → Tin tức về topic
   • Tool: summarize_article(url: str) → Tóm tắt bài

3. Personal Notes Server

   • Tool: add_note(content: str) → Lưu note
   • Tool: search_notes(query: str) → Tìm notes
   • Resource: notes://recent → 10 notes gần nhất

Template

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()

---

Tóm tắt

Bước	Action
1	Cài pip install mcp
2	Viết @mcp.tool() functions
3	Test với Inspector
4	Config Claude Desktop
5	Restart và test

Chúc mừng! Bạn đã hoàn thành chuyên mục AI Engineering! 🎉