从零实现一个MCP Server：让你的Agent操控数据库、搜索引擎和API 🛠️

导读：上一期我们了解了MCP协议的原理，这期直接动手！从零开始实现一个功能完整的MCP Server，让你的Agent能够操控数据库、调用搜索引擎、访问外部API。跟着做，30分钟搞定！💪

---

一、本期目标 🎯

我们要实现一个全能MCP Server，包含以下能力：

功能模块	工具	说明
🗄️ 数据库	query_db, list_tables, describe_table	查询数据库
🔍 搜索	web_search, search_news	网络搜索
📁 文件	read_file, write_file, list_dir	文件操作
📊 数据	create_chart, export_csv	数据可视化

架构图

---

二、项目结构 📂

mcp_super_server/
├── server.py          # MCP Server主文件
├── tools/
│   ├── database.py    # 数据库工具
│   ├── search.py      # 搜索工具
│   ├── filesystem.py  # 文件工具
│   └── chart.py       # 图表工具
├── config.py          # 配置文件
└── requirements.txt   # 依赖

---

三、完整代码实现 💻

3.1 主Server文件

"""
server.py - 全能MCP Server
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource
import asyncio
import json

# 创建Server
server = Server("super-tool-server")

# ===== 注册所有工具 =====

# --- 数据库工具 ---
@server.tool()
async def query_database(sql: str, db_path: str = "data.db") -> list[TextContent]:
    """执行SQL查询（仅支持SELECT）。用于查询数据库中的数据。
    
    Args:
        sql: SQL查询语句，如 'SELECT * FROM users LIMIT 10'
        db_path: 数据库文件路径，默认data.db
    """
    import sqlite3
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
        conn.close()
        
        result = json.dumps({
            "success": True,
            "columns": columns,
            "data": rows,
            "total": len(rows)
        }, ensure_ascii=False, indent=2)
        return [TextContent(type="text", text=result)]
    except Exception as e:
        return [TextContent(type="text", text=json.dumps({"success": False, "error": str(e)}))]

@server.tool()
async def list_tables(db_path: str = "data.db") -> list[TextContent]:
    """列出数据库中的所有表。
    
    Args:
        db_path: 数据库文件路径
    """
    import sqlite3
    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 [TextContent(type="text", text=json.dumps({"tables": tables}))]

@server.tool()
async def describe_table(table_name: str, db_path: str = "data.db") -> list[TextContent]:
    """获取表的结构信息（列名、类型等）。
    
    Args:
        table_name: 表名
        db_path: 数据库文件路径
    """
    import sqlite3
    conn = sqlite3.connect(db_path)
    cursor = conn.execute(f"PRAGMA table_info({table_name})")
    columns = [{"cid": r[0], "name": r[1], "type": r[2], "notnull": r[3], "default": r[4], "pk": r[5]} 
               for r in cursor.fetchall()]
    conn.close()
    return [TextContent(type="text", text=json.dumps({"table": table_name, "columns": columns}, ensure_ascii=False))]

# --- 搜索工具 ---
@server.tool()
async def web_search(query: str, max_results: int = 5) -> list[TextContent]:
    """搜索互联网获取信息。当需要查询实时数据、新闻或不确定的事实时使用。
    
    Args:
        query: 搜索关键词
        max_results: 最大结果数，默认5
    """
    # 实际项目中接入Tavily/Bing等搜索API
    # 这里展示接口结构
    results = []
    # ... 实际搜索逻辑 ...
    return [TextContent(type="text", text=json.dumps({
        "query": query,
        "results": results,
        "note": "请配置搜索API Key以启用真实搜索"
    }, ensure_ascii=False))]

# --- 文件工具 ---
@server.tool()
async def read_file(file_path: str) -> list[TextContent]:
    """读取文件内容。支持文本文件。
    
    Args:
        file_path: 文件路径
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        return [TextContent(type="text", text=json.dumps({
            "success": True, "path": file_path, "content": content,
            "size": len(content)
        }, ensure_ascii=False))]
    except Exception as e:
        return [TextContent(type="text", text=json.dumps({"success": False, "error": str(e)}))]

@server.tool()
async def write_file(file_path: str, content: str) -> list[TextContent]:
    """写入内容到文件。
    
    Args:
        file_path: 文件路径
        content: 要写入的内容
    """
    try:
        import os
        os.makedirs(os.path.dirname(file_path) or '.', exist_ok=True)
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(content)
        return [TextContent(type="text", text=json.dumps({
            "success": True, "path": file_path, "bytes_written": len(content)
        }))]
    except Exception as e:
        return [TextContent(type="text", text=json.dumps({"success": False, "error": str(e)}))]

@server.tool()
async def list_directory(dir_path: str = ".") -> list[TextContent]:
    """列出目录下的文件和子目录。
    
    Args:
        dir_path: 目录路径，默认当前目录
    """
    import os
    try:
        items = []
        for item in os.listdir(dir_path):
            full_path = os.path.join(dir_path, item)
            items.append({
                "name": item,
                "type": "directory" if os.path.isdir(full_path) else "file",
                "size": os.path.getsize(full_path) if os.path.isfile(full_path) else None
            })
        return [TextContent(type="text", text=json.dumps({"path": dir_path, "items": items}, ensure_ascii=False))]
    except Exception as e:
        return [TextContent(type="text", text=json.dumps({"success": False, "error": str(e)}))]

# --- 图表工具 ---
@server.tool()
async def create_chart(chart_type: str, data: str, title: str = "Chart", output_path: str = "chart.png") -> list[TextContent]:
    """生成数据图表。支持折线图、柱状图、饼图。
    
    Args:
        chart_type: 图表类型，'line'(折线图)、'bar'(柱状图)、'pie'(饼图)
        data: JSON格式的数据，如 '{"labels":["A","B","C"],"values":[10,20,30]}'
        title: 图表标题
        output_path: 输出图片路径
    """
    import matplotlib.pyplot as plt
    import matplotlib
    matplotlib.use('Agg')
    
    chart_data = json.loads(data)
    labels = chart_data.get("labels", [])
    values = chart_data.get("values", [])
    
    fig, ax = plt.subplots(figsize=(10, 6))
    
    if chart_type == "line":
        ax.plot(labels, values, marker='o', linewidth=2)
    elif chart_type == "bar":
        ax.bar(labels, values, color='skyblue')
    elif chart_type == "pie":
        ax.pie(values, labels=labels, autopct='%1.1f%%')
    
    ax.set_title(title)
    plt.tight_layout()
    plt.savefig(output_path, dpi=150)
    plt.close()
    
    return [TextContent(type="text", text=json.dumps({
        "success": True, "path": output_path, "type": chart_type
    }))]

# ===== 注册资源 =====
@server.resource("config://server-info")
async def get_server_info() -> str:
    """获取Server信息"""
    return json.dumps({
        "name": "Super Tool Server",
        "version": "1.0.0",
        "tools": ["query_database", "list_tables", "describe_table", 
                  "web_search", "read_file", "write_file", "list_directory",
                  "create_chart"]
    }, ensure_ascii=False)

# ===== 启动 =====
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

if __name__ == "__main__":
    asyncio.run(main())

---

四、测试MCP Server 🧪

4.1 使用MCP Inspector测试

# 安装MCP Inspector（官方调试工具）
npx @modelcontextprotocol/inspector python server.py

MCP Inspector会打开一个Web界面，你可以：

• 📋 查看所有注册的工具
• 🧪 手动调用工具测试
• 📊 查看返回结果

4.2 使用Python Client测试

"""test_client.py - 测试MCP Server"""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def test():
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 列出所有工具
            tools = await session.list_tools()
            print("📋 可用工具：")
            for tool in tools.tools:
                print(f"  🔧 {tool.name}: {tool.description[:50]}...")
            
            # 测试数据库查询
            result = await session.call_tool("list_tables", {"db_path": "test.db"})
            print(f"\n🗄️ 数据库表：{result.content[0].text}")
            
            # 测试文件操作
            result = await session.call_tool("write_file", {
                "file_path": "/tmp/test.txt",
                "content": "Hello MCP!"
            })
            print(f"\n📁 写入文件：{result.content[0].text}")
            
            # 测试图表生成
            import json
            chart_data = json.dumps({
                "labels": ["Jan", "Feb", "Mar", "Apr", "May"],
                "values": [100, 200, 150, 300, 250]
            })
            result = await session.call_tool("create_chart", {
                "chart_type": "bar",
                "data": chart_data,
                "title": "Monthly Sales",
                "output_path": "/tmp/sales_chart.png"
            })
            print(f"\n📊 生成图表：{result.content[0].text}")

asyncio.run(test())

---

五、在Agent中集成MCP Server 🤖

5.1 完整的Agent + MCP集成

"""
agent_with_mcp.py - Agent集成MCP Server
"""
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def main():
    # 连接MCP Server
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 加载MCP工具
            mcp_tools = await load_mcp_tools(session)
            
            # 创建Agent
            agent = create_react_agent(
                model=ChatOpenAI(model="gpt-4o", temperature=0),
                tools=mcp_tools,
                prompt="""你是一个全能AI助手，拥有以下能力：
                - 🗄️ 查询数据库（query_database, list_tables, describe_table）
                - 🔍 搜索互联网（web_search）
                - 📁 读写文件（read_file, write_file, list_directory）
                - 📊 生成图表（create_chart）
                
                根据用户需求，灵活组合使用这些工具。""",
                checkpointer=MemorySaver()
            )
            
            # 交互式对话
            config = {"configurable": {"thread_id": "demo-1"}}
            
            while True:
                user_input = input("\n你：")
                if user_input.lower() in ["exit", "quit"]:
                    break
                
                result = await agent.ainvoke(
                    {"messages": [("user", user_input)]},
                    config=config
                )
                print(f"\n助手：{result['messages'][-1].content}")

asyncio.run(main())

5.2 测试效果

你：帮我看看数据库里有什么表？
助手：🗄️ [调用list_tables工具]
      数据库中有以下表：
      - users（用户表）
      - orders（订单表）
      - products（产品表）

你：查一下用户表的前5条数据
助手：🗄️ [调用query_database工具]
      用户表前5条数据：
      | ID | 姓名 | 邮箱 | 注册时间 |
      |----|------|------|----------|
      | 1  | 张三 | zhang@test.com | 2025-01-01 |
      ...

你：把这些数据生成一个柱状图
助手：📊 [调用create_chart工具]
      图表已生成并保存到 chart.png！

你：帮我写一份数据分析报告，保存到report.md
助手：📁 [调用write_file工具]
      报告已保存到 report.md！包含以下内容：
      1. 数据概览
      2. 用户分析
      3. 趋势图表
      ...

---

六、MCP Server最佳实践 📋

6.1 开发规范

规范	说明	示例
📝 描述要详细	工具描述要清晰，包含使用场景	“当用户需要查询…时使用”
📋 参数要验证	检查参数合法性	SQL只允许SELECT
🔄 错误要处理	捕获异常返回友好信息	返回JSON格式的错误信息
🔒 安全要保障	限制危险操作	禁止DELETE/DROP
📊 日志要记录	记录工具调用日志	方便调试和审计

6.2 性能优化

优化项	方法	效果
🔗 连接池	数据库连接复用	响应速度↑50%
🗂️ 缓存	频繁查询结果缓存	重复请求零延迟
⚡ 并行执行	多个工具并行调用	总耗时↓
📦 分页返回	大数据集分页	避免内存溢出

---

七、本期小结 📝

内容	要点
MCP Server开发	使用@server.tool()装饰器，几十行代码搞定
工具类型	数据库查询、搜索、文件操作、图表生成
测试方法	MCP Inspector（官方）或Python Client
Agent集成	用langchain_mcp_adapters一行代码加载工具
最佳实践	描述详细、参数验证、错误处理、安全保障

🔥 MCP Server开发真的非常简单！ 难的不是写代码，而是设计好工具的接口和描述，让Agent能准确地选择和使用你的工具。

---

📢 下期预告：《Agent工具链实战：搜索+代码执行+文件操作，打造全能工具箱》—— 我们继续丰富Agent的工具箱，让它成为一个真正的全能助手！🧰

---

📌 三连走起！动手写一个MCP Server，让你的Agent更强大！ 💪

📚 专栏第10/24期，工具与协议篇进行中…

作者：高炉炼铁智能化技术研究者，专注钢铁冶金与人工智能 交叉领域。

👍 如果觉得有帮助，请点赞、收藏、转发！
版权归作者所有，未经许可请勿抄袭，套用，商用(或其它具有利益性行为)。
🔔 关注专栏，不错过后续精彩内容