llm

MCP 服务端开发实战：从 SQLite 查询服务到文件系统与 GitHub API 完整实现

手把手实现 3 个生产级 MCP Server：SQLite 数据库查询、文件系统操作（安全沙箱）、GitHub API 集成。涵盖 Server 生命周期管理（初始化 / tool 发现 / 权限声明）、错误处理、连接安全、性能调优。附完整的 Python 实现代码、单元测试用例与 Docker 部署方案。

Leeting Yan 2026-08-08 5 分钟阅读 2197 字

前置知识：MCP 协议概览，了解 Resources / Prompts / Tools / Sampling 四大原语。

1. MCP Server 开发全景图

开发流程
├── Step 1: 明确 Server 暴露的能力集合
│   ├── Resources（只读数据）→ SQL 查询结果、文件内容
│   ├── Tools（可写操作）→ UPDATE、INSERT、删除文件
│   └── Prompts（复用模板）→ 标准化的分析 Prompt
│
├── Step 2: 选择传输层
│   ├── stdio（本地脚本）→ 50% 场景
│   ├── SSE（HTTP 服务）→ 30% 场景
│   └── WebSocket（实时）→ 20% 场景
│
├── Step 3: 实现生命周期回调
│   ├── list_resources / read_resource
│   ├── list_tools / call_tool
│   └── list_prompts / get_prompt
│
├── Step 4: 安全加固
│   ├── 文件路径白名单（防目录遍历）
│   ├── SQL 只读限制（防 DROP TABLE）
│   └── 请求签名 / API Key 验证（远程 Server）
│
└── Step 5: 测试与部署
    ├── 单元测试（Mock MCP Client）
    ├── 集成测试（连接真实 LLM）
    └── Docker 封装（一键部署）

2. 实战一：SQLite MCP Server

2.1 安装依赖

pip install mcp aiosqlite

2.2 实现代码

import asyncio, aiosqlite
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.server.models import InitializationOptions
import mcp.types as types
from pathlib import Path

class SQLiteMCPServer:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.server = Server("sqlite-mcp")
        self._setup_handlers()

    def _setup_handlers(self):
        @self.server.list_resources()
        async def list_resources() -> list[types.Resource]:
            async with aiosqlite.connect(self.db_path) as db:
                cursor = await db.execute(
                    "SELECT name FROM sqlite_master WHERE type='table'"
                )
                tables = await cursor.fetchall()
                return [
                    types.Resource(
                        uri=f"sqlite:///{table[0]}/schema",
                        name=table[0],
                        mimeType="application/json",
                        description=f"表 {table[0]} 的 Schema",
                    )
                    for table in tables
                ]

        @self.server.read_resource()
        async def read_resource(uri: str) -> str:
            if not uri.startswith("sqlite:///"):
                raise ValueError(f"不支持的 URI 格式: {uri}")
            table_name = uri.replace("sqlite:///", "").replace("/schema", "")
            async with aiosqlite.connect(self.db_path) as db:
                cursor = await db.execute(f"PRAGMA table_info({table_name})")
                cols = await cursor.fetchall()
                return str([{"name": c[1], "type": c[2]} for c in cols])

        @self.server.list_tools()
        async def list_tools() -> list[types.Tool]:
            return [
                types.Tool(
                    name="query",
                    description="对数据库执行安全的只读 SELECT 查询",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "sql": {
                                "type": "string",
                                "description": "SQL SELECT 语句（只读）",
                            }
                        },
                        "required": ["sql"],
                    },
                ),
                types.Tool(
                    name="execute",
                    description="执行 INSERT/UPDATE/DELETE（写权限）",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "sql": {"type": "string", "description": "SQL 写语句"}
                        },
                        "required": ["sql"],
                    },
                ),
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
            sql = arguments.get("sql", "")

            # 安全检查：READ 操作只允许 SELECT
            if name == "query":
                if not sql.strip().lower().startswith("select"):
                    return [types.TextContent(
                        type="text",
                        text="⚠️ 只读查询仅支持 SELECT 语句，如需写入请使用 execute 工具。"
                    )]
                async with aiosqlite.connect(self.db_path) as db:
                    db.row_factory = aiosqlite.Row
                    cursor = await db.execute(sql)
                    rows = await cursor.fetchall()
                    dicts = [dict(row) for row in rows]
                    return [types.TextContent(type="text", text=str(dicts))]

            # WRITE 操作需额外确认（本示例简化）
            elif name == "execute":
                async with aiosqlite.connect(self.db_path) as db:
                    await db.execute(sql)
                    await db.commit()
                    return [types.TextContent(type="text", text="✅ 执行成功")]

            raise ValueError(f"未知工具: {name}")

    async def run(self):
        async with stdio_server(self.server) as (read_stream, write_stream):
            await self.server.run(
                read_stream, write_stream,
                InitializationOptions(
                    server_name="sqlite-mcp",
                    server_version="1.0.0",
                    capabilities=self.server.get_capabilities(),
                ),
            )

if __name__ == "__main__":
    import sys
    db_path = sys.argv[1] if len(sys.argv) > 1 else "./database.db"
    server = SQLiteMCPServer(db_path)
    asyncio.run(server.run())

2.3 配置到 Claude Desktop

{
  "mcpServers": {
    "my-database": {
      "command": "python",
      "args": ["/path/to/sqlite_mcp.py", "./prod.db"]
    }
  }
}

2.4 测试

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' \
  | python sqlite_mcp.py ./test.db

3. 实战二：文件系统 MCP Server（安全沙箱版）

from pathlib import Path
from typing import Sequence
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
import asyncio

class FileSystemMCPServer:
    def __init__(self, root_path: str):
        self.root = Path(root_path).resolve()
        self.server = Server("fs-mcp")
        self._setup()

    def _is_safe(self, path: Path) -> bool:
        try:
            resolved = path.resolve()
            return str(resolved).startswith(str(self.root))
        except:
            return False

    def _setup(self):
        @self.server.list_tools()
        async def list_tools() -> list[types.Tool]:
            return [
                types.Tool(
                    name="read_file",
                    description="读取沙箱内文件内容",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "description": "相对路径"}
                        },
                        "required": ["path"],
                    },
                ),
                types.Tool(
                    name="write_file",
                    description="写入沙箱内文件（覆盖模式）",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string"},
                            "content": {"type": "string"},
                        },
                        "required": ["path", "content"],
                    },
                ),
                types.Tool(
                    name="list_dir",
                    description="列出沙箱目录内容",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "default": "."}
                        },
                    },
                ),
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
            target = self.root / arguments.get("path", ".")
            if not self._is_safe(target):
                return [types.TextContent(type="text", text="❌ 访问超出沙箱范围")]

            if name == "read_file":
                content = target.read_text(encoding="utf-8")
                return [types.TextContent(type="text", text=content)]

            elif name == "write_file":
                target.write_text(arguments["content"], encoding="utf-8")
                return [types.TextContent(type="text", text="✅ 写入成功")]

            elif name == "list_dir":
                entries = [p.name for p in target.iterdir()]
                return [types.TextContent(type="text", text="\n".join(entries))]

            raise ValueError(f"未知工具: {name}")

    async def run(self):
        async with stdio_server(self.server) as (rs, ws):
            await self.server.run(rs, ws, InitializationOptions(
                server_name="fs-mcp", server_version="1.0.0",
                capabilities=self.server.get_capabilities()
            ))

if __name__ == "__main__":
    import sys
    root = sys.argv[1] if len(sys.argv) > 1 else "./workspace"
    asyncio.run(FileSystemMCPServer(root).run())

4. 实战三：GitHub API MCP Server（远程 SSE 传输）

from mcp.server import Server
from mcp.server.sse import SseServerTransport
import mcp.types as types
from starlette.applications import Starlette
from starlette.routing import Route
import httpx
import os

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")

async def github_mcp_server(scope, receive, send):
    app = Server("github-mcp")

    @app.list_tools()
    async def list_tools() -> list[types.Tool]:
        return [
            types.Tool(
                name="search_repos",
                description="搜索 GitHub 仓库",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "language": {"type": "string"},
                    },
                    "required": ["query"],
                },
            ),
            types.Tool(
                name="get_repo_info",
                description="获取仓库详情",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "owner": {"type": "string"},
                        "repo": {"type": "string"},
                    },
                    "required": ["owner", "repo"],
                },
            ),
        ]

    @app.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
        async with httpx.AsyncClient() as client:
            if name == "search_repos":
                q = arguments["query"]
                lang = arguments.get("language", "")
                query = f"{q} language:{lang}" if lang else q
                resp = await client.get(
                    f"https://api.github.com/search/repositories?q={query}",
                    headers={"Authorization": f"token {GITHUB_TOKEN}"},
                )
                data = resp.json()
                repos = [r["full_name"] for r in data.get("items", [])[:5]]
                return [types.TextContent(type="text", text="\n".join(repos))]

            elif name == "get_repo_info":
                resp = await client.get(
                    f"https://api.github.com/repos/{arguments['owner']}/{arguments['repo']}",
                    headers={"Authorization": f"token {GITHUB_TOKEN}"},
                )
                return [types.TextContent(type="text", text=resp.text)]

    # SSE 传输适配
    transport = SseServerTransport("/messages")
    return await transport.connect(scope, receive, send)

app = Starlette(routes=[
    Route("/sse", github_mcp_server),
])

启动：uvicorn github_mcp:app --host 0.0.0.0 --port 8000

5. 安全加固清单

□ 输入校验：所有路径参数 resolve() 后检查是否超出沙箱
□ SQL 限制：query 只允许 SELECT，execute 需人工审批
□ API 认证：远程 Server 使用 API Key / Token
□ 速率限制：每个客户端每分钟最多 60 次调用
□ 超时控制：单次工具调用最多 30 秒
□ 审计日志：记录调用者、参数、结果、耗时
□ 错误脱敏：生产环境不暴露栈跟踪给用户
□ 最小权限：Server 进程使用非 root 账号运行

6. 测试策略

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_sqlite_readonly_guard():
    # Mock MCP Client 调用
    from sqlite_mcp import SQLiteMCPServer
    server = SQLiteMCPServer(":memory:")

    # 尝试用 query 工具执行 DROP → 应被拒绝
    result = await server.server.call_tool("query", {"sql": "DROP TABLE users"})
    assert "⚠️" in result[0].text

@pytest.mark.asyncio
async def test_filesystem_sandbox():
    from fs_mcp import FileSystemMCPServer
    server = FileSystemMCPServer("/tmp/safe")

    # 尝试访问 /etc/passwd → 应被拒绝
    result = await server.server.call_tool("read_file", {"path": "../../etc/passwd"})
    assert "❌" in result[0].text

7. Docker 部署

FROM python:3.12-slim

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

COPY *.py .
EXPOSE 8000

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

FAQ

Q: stdio 和 SSE 两种传输方式怎么选？
A: 如果 Server 和 Client 在同一台机器、不需要网络隔离，用 stdio（最简单）。如果需要远程访问、多客户端共享、或部署到容器/云环境，用 SSE。

Q: MCP Server 需要认证吗？
A: stdio 模式下由操作系统进程隔离，通常不需要。SSE 模式下强烈推荐 API Key / OAuth2 认证。

Q: 怎么调试 MCP Server？
A: 1) 先用 echo + stdin 做协议级调试；2) 用 Claude Desktop 的 MCP Inspector 查看工具列表；3) 查看 Server 的 stderr 日志。

📂 继续阅读：
MCP 协议概览 — 四大原语与架构全景
MCP 客户端集成 — Claude Desktop、Cline、Continue 配置
MCP 工具生态 — 20+ Server 速查表

← 上一篇 人机协作模式（Human-in-the-Loop）：让 Agent 在关键时刻停下来，等待人类确认 下一篇 → MCP 工具生态与最佳实践：30+ Server 速查表与场景化选型指南

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页