Python中实现MCP协议的开发指南

什么是 MCP 开发?为何开发者需要关注?

MCP(Model Context Protocol)是一项革命性技术,它为 AI 应用提供了标准化接口,让大语言模型(LLM)能够轻松访问外部工具、数据和系统。想象一下 USB 如何标准化了设备连接,MCP 就是 AI 世界的"万能接口"。

对开发者而言,MCP 解决了 AI 集成的最大痛点:如何让 AI 与你的系统安全、高效地交互。无需复杂的自定义接口,MCP 提供了一套统一标准,显著降低了开发成本和技术门槛。

MCP 开发能解决哪些实际问题?

1. 数据孤岛问题

传统 AI 只能处理对话框内的信息,而 MCP 让 AI 能够:

  • 直接查询你的数据库
  • 访问内部知识库
  • 读取本地文件系统
  • 连接私有 API
  • 2. 工具调用壁垒

    通过 MCP,AI 可以:

  • 执行代码并返回结果
  • 调用企业内部系统
  • 操作第三方服务
  • 控制本地应用程序
  • 3. 安全与控制挑战

    MCP 提供:

  • 细粒度的权限控制
  • 人工审核机制
  • 操作审计日志
  • 数据本地处理能力
  • 实战案例:用 Python 开发 MCP 服务器

    案例一:企业知识库助手

    # knowledge_base.py
    from mcp.server.fastmcp import FastMCP
    import sqlite3
    
    mcp = FastMCP("knowledge-base")
    
    @mcp.tool()
    async def search_documents(query: str, department: str = None) -> str:
       """搜索企业文档库
    
       Args:
           query: 搜索关键词
           department: 可选的部门筛选
       """
       conn = sqlite3.connect('documents.db')
       cursor = conn.cursor()
    
       if department:
           sql = "SELECT title, summary FROM documents WHERE content LIKE ? AND department = ?"
           cursor.execute(sql, (f"%{query}%", department))
       else:
           sql = "SELECT title, summary FROM documents WHERE content LIKE ?"
           cursor.execute(sql, (f"%{query}%",))
    
       results = cursor.fetchall()
       conn.close()
    
       if not results:
           return f"未找到与'{query}'相关的文档" # Changed double quotes to single quotes for clarity within f-string
    
       return "\n\n".join([f"📄 {title}\n{summary}" for title, summary in results[:5]])
    
    if __name__ == "__main__":
       mcp.run(transport='stdio')
    

    有了这个 MCP 服务器,用户可以询问:

    "帮我找一下市场部关于第四季度营销策略的文档" "我需要技术团队关于云迁移的最新报告"

    AI 将直接从企业知识库中检索相关信息,而不是返回"我无法访问您的内部系统"这样的回答。

    案例二:全栈开发助手

    # dev_assistant.py
    from mcp.server.fastmcp import FastMCP
    import subprocess
    import os
    
    mcp = FastMCP("dev-tools")
    
    @mcp.tool()
    async def run_code(language: str, code: str) -> str:
        """执行代码并返回结果
    
        Args:
            language: 编程语言(python, javascript, bash)
            code: 要执行的代码
        """
        if language == "python":
            # 创建临时文件
            with open("temp.py", "w") as f:
                f.write(code)
    
            # 执行代码并捕获输出
            result = subprocess.run(
                ["python", "temp.py"],
                capture_output=True,
                text=True
            )
    
            # 清理临时文件
            os.remove("temp.py")
    
            # 返回执行结果
            if result.returncode == 0:
                return f"执行成功:\n{result.stdout}"
            else:
                return f"执行出错:\n{result.stderr}"
    
        elif language == "javascript":
            # 类似实现...
            return "JavaScript 执行暂未实现" # Added placeholder return
    
        elif language == "bash":
            # 类似实现...
            result = subprocess.run(
                code,
                shell=True, # Use shell=True for bash commands carefully
                capture_output=True,
                text=True
            )
            if result.returncode == 0:
                return f"执行成功:\n{result.stdout}"
            else:
                return f"执行出错:\n{result.stderr}"
    
        else:
            return f"不支持的语言:{language}"
    
    @mcp.tool()
    async def search_codebase(query: str, file_type: str = None) -> str:
        """搜索代码库
    
        Args:
            query: 搜索关键词
            file_type: 可选的文件类型筛选(如 py, js, css)
        """
        cmd = ["grep", "-r", query, "."]
        if file_type:
            cmd = ["grep", "-r", query, "--include", f"*.{file_type}", "."]
    
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, check=False) # Use check=False to handle no results
    
            if not result.stdout:
                return f"未找到包含'{query}'的代码" # Changed double quotes to single quotes
    
            # 限制返回结果数量,避免过多
            lines = result.stdout.strip().split("\n")[:10]
            return "\n".join(lines)
        except FileNotFoundError:
            return "错误:未找到 grep 命令。请确保它已安装并在 PATH 中。"
        except Exception as e:
            return f"搜索代码库时出错: {str(e)}"
    
    
    if __name__ == "__main__":
        mcp.run(transport='stdio')
    

    借助这个 MCP 服务器,开发者可以让 AI 帮助:

  • 执行测试代码片段并分析结果
  • 在代码库中查找特定功能的实现
  • 生成并立即测试解决方案
  • 案例三:智能数据分析助手

    # data_analyst.py
    from mcp.server.fastmcp import FastMCP
    import pandas as pd
    import matplotlib.pyplot as plt
    import io
    import base64
    import os # Import os
    
    mcp = FastMCP("data-tools")
    
    @mcp.tool()
    async def analyze_csv(filepath: str, analysis_type: str) -> str:
        """分析 CSV 数据文件
    
        Args:
            filepath: CSV 文件路径
            analysis_type: 分析类型(summary, correlation, group_by)
        """
        try:
            # 读取 CSV 文件
            df = pd.read_csv(filepath)
    
            if analysis_type == "summary":
                # 返回数据摘要
                summary = df.describe().to_string()
                return f"数据摘要:\n{summary}"
    
            elif analysis_type == "correlation":
                # 计算相关性 (only on numeric columns)
                numeric_df = df.select_dtypes(include='number')
                if numeric_df.empty:
                    return "文件中没有数值列,无法计算相关性。"
                corr = numeric_df.corr().to_string()
                return f"相关性分析:\n{corr}"
    
            elif analysis_type == "group_by":
                # 假设文件中有 category 列
                if "category" in df.columns:
                    # Group by 'category' and calculate mean for numeric columns
                    grouped = df.groupby("category").agg({col: 'mean' for col in df.select_dtypes(include='number').columns}).to_string()
                    return f"分组统计 (均值):\n{grouped}"
                else:
                    return "文件中没有 category 列,无法进行分组分析"
    
            else:
                return f"不支持的分析类型:{analysis_type}"
        except FileNotFoundError:
            return f"错误:文件未找到 '{filepath}'"
        except Exception as e:
            return f"分析 CSV 时出错: {str(e)}"
    
    @mcp.tool()
    async def generate_chart(filepath: str, chart_type: str, x_column: str, y_column: str) -> str:
        """生成数据可视化图表
    
        Args:
            filepath: CSV 文件路径
            chart_type: 图表类型(line, bar, scatter)
            x_column: X 轴列名
            y_column: Y 轴列名
        """
        try:
            # 读取 CSV 文件
            df = pd.read_csv(filepath)
    
            # 检查列是否存在
            if x_column not in df.columns or y_column not in df.columns:
                return f"错误:列 '{x_column}' 或 '{y_column}' 不在文件中。"
    
            # 创建图表
            plt.figure(figsize=(10, 6))
    
            if chart_type == "line":
                plt.plot(df[x_column], df[y_column])
            elif chart_type == "bar":
                # Bar chart often needs aggregation or specific handling if x_column is not unique categories
                # For simplicity, assuming x_column represents categories or distinct values
                plt.bar(df[x_column], df[y_column])
            elif chart_type == "scatter":
                plt.scatter(df[x_column], df[y_column])
            else:
                plt.close() # Close the figure if chart type is invalid
                return f"不支持的图表类型:{chart_type}"
    
            plt.xlabel(x_column)
            plt.ylabel(y_column)
            plt.title(f"{y_column} vs {x_column} ({chart_type.capitalize()} Chart)")
            plt.tight_layout() # Adjust layout
    
            # 将图表保存为图片文件
            img_path = "chart.png"
            plt.savefig(img_path)
            plt.close() # Close the plot figure
    
            return f"图表已生成并保存为:{img_path}"
        except FileNotFoundError:
            return f"错误:文件未找到 '{filepath}'"
        except Exception as e:
            plt.close() # Ensure figure is closed on error
            return f"生成图表时出错: {str(e)}"
    
    
    if __name__ == "__main__":
        mcp.run(transport='stdio')
    

    数据分析师可以请求:

    "帮我分析这个销售数据的相关性" "根据产品类别生成一个条形图,展示各类别的销售额" "计算并展示客户年龄与购买金额的关系"

    MCP 开发的核心优势

    1. 降低集成成本 传统 AI 集成需要定制开发,而 MCP 提供即插即用的体验,大幅降低工程成本。一个 MCP 服务器可以同时为多个 AI 客户端提供服务,实现一次开发多处使用。
    2. 增强 AI 能力 通过 MCP,AI 能够突破对话框限制,实现与真实系统的深度交互。不再仅限于回答问题,AI 可以执行任务、分析数据、控制系统。
    3. 安全与控制 MCP 设计中包含人工审核机制,确保 AI 的每一个关键操作都经过用户确认。数据可以在本地处理,无需上传到云端,提供更高的数据安全保障。
    4. 跨平台兼容 MCP 是厂商中立的开放协议,你的 MCP 服务器可以无缝连接不同的 AI 提供商,避免被单一平台锁定。

    入门 MCP 开发的简单步骤

    安装必要工具:

    # 使用 uv 安装(推荐)
    uv pip install mcp httpx
    
    # 或使用传统 pip
    pip install mcp httpx
    

    创建你的第一个 MCP 服务器:

    # hello_mcp.py
    from mcp.server.fastmcp import FastMCP
    
    # 初始化 MCP 服务器
    mcp = FastMCP("hello-world")
    
    @mcp.tool()
    async def greet(name: str) -> str:
        """向用户问好
    
        Args:
            name: 用户名称
        """
        return f"你好,{name}!欢迎使用 MCP!"
    
    @mcp.tool()
    async def calculate(expression: str) -> str:
        """计算数学表达式 (注意: eval() 有安全风险,仅用于演示)
    
        Args:
            expression: 数学表达式
        """
        try:
            # 警告: eval() 可以执行任意代码,非常不安全。
            # 在生产环境中,应使用更安全的表达式求值库 (如 numexpr, ast.literal_eval)
            allowed_chars = "0123456789+-*/(). "
            if not all(c in allowed_chars for c in expression):
                 return "计算出错: 表达式包含不允许的字符"
            result = eval(expression)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算出错: {str(e)}"
    
    if __name__ == "__main__":
        # 启动服务器
        mcp.run(transport='stdio')
    

    运行并测试:

    python hello_mcp.py
    

    与 Claude Desktop 集成:

    编辑 Claude Desktop 配置文件(macOS: ~/Library/Application Support/Claude/claude_desktop_config.json 或 Windows: %APPDATA%\Claude\claude_desktop_config.json),添加:

    {
      "mcpServers": {
        "hello-world": {
          "command": "python",
          "args": ["/完整/路径/到/hello_mcp.py"]
        }
      }
    }
    

    适合 MCP 开发的场景

    MCP 开发特别适合这些场景:

  • 企业内部系统集成:连接 CRM、ERP、知识库等内部系统
  • 开发辅助工具:代码分析、生成、测试和文档工具
  • 数据分析应用:连接数据库和分析工具
  • 物联网控制中心:管理和控制智能设备
  • 个人生产力助手:文件管理、日程安排、邮件处理
  • 结语

    MCP 开发正在改变 AI 应用的格局,它使 AI 从简单的对话机器人转变为能够实际行动和解决问题的助手。作为开发者,掌握 MCP 开发不仅能提升你的技术栈,还能为你的团队或客户创造真正有价值的 AI 解决方案。

    无论你是希望增强现有系统的智能化水平,还是构建全新的 AI 驱动应用,MCP 都提供了一条清晰、高效的技术路径。开始 MCP 开发,探索 AI 与实际系统无缝融合的新可能!

    README.md · randomuser1123/MCP开发指南 – Gitee.com

    作者:m0_62919969

    物联沃分享整理
    物联沃-IOTWORD物联网 » Python中实现MCP协议的开发指南

    发表回复