AI导哈

MCP服务器构建工具

一个使用Python构建模型上下文协议(MCP)服务器的工具，用于连接AI应用与外部数据源和工具。

  • by pavantata09
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 pavantata09/MCP-tutorial

使用说明

MCP服务器教程-Python

使用Python构建模型上下文协议（MCP）服务器的综合指南。

什么是MCP？

模型上下文协议（MCP）是Anthropic开发的一个开放标准，它使人工智能应用程序，特别是大型语言模型（LLM），能够安全地与外部数据源和工具连接。将其视为一个通用接口，就像人工智能的USB-C端口一样，允许人工智能模型和应用程序之间的无缝连接。

为什么MCP很重要

MCP解决了LLM与外部数据和工具隔离的问题。MCP没有为每个工具或数据源构建自定义集成，而是提供：

  • 标准化接口：一种将人工智能与多种工具和数据源连接起来的协议
  • 安全连接：访问外部资源的内置安全机制
  • 实时数据访问：使AI能够从API、数据库和文件中获取实时数据
  • 扩展功能：让AI执行发送电子邮件、查询数据库或执行代码等操作

先决条件

在开始之前，请确保您已经：

  • 已安装Python 3.9或更高版本
  • Python编程的基本理解
  • pip包管理器
  • 文本编辑器或IDE

安装

1.设置您的环境

创建一个虚拟环境来隔离您的依赖关系：

python -m venv mcp-env
source mcp-env/bin/activate  # On Windows: mcp-env\Scripts\activate

2.安装所需的软件包

安装FastMCP库（简化的MCP服务器框架）：

pip install mcp fastmcp

或者使用CLI工具进行安装：

pip install "mcp[cli]"

根据您的用例，您可能还需要其他库：

pip install requests  # For HTTP calls

构建您的第一台MCP服务器

基本服务器结构

创建一个名为的文件 server.py:

from fastmcp import FastMCP

# Create a server instance with a descriptive name
mcp = FastMCP(name="My First MCP Server")

# Define a simple tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Adds two integer numbers together."""
    return a + b

# Make the server runnable
if __name__ == "__main__":
    mcp.run()

关键部件说明

  1. 服务器实例: FastMCP(name="...") 使用唯一名称创建MCP服务器
  2. 工具：功能装饰 @mcp.tool() AI可以执行
  3. 类型提示：Python类型注释会自动为输入生成JSON模式
  4. 文档字符串：函数文档字符串成为AI的工具描述

FastMCP的工作原理

FastMCP自动处理复杂性：

  • 工具名称：使用函数名作为工具的标识符
  • 描述：提取文档字符串作为工具描述
  • 模式：从类型提示生成JSON架构
  • 注册：为AI模型提供工具

项目结构

对于更复杂的服务器，请使用此组织：

my_mcp_server/
│
├── data/              # Sample data files
├── tools/             # MCP tool definitions
├── utils/             # Reusable utility functions
├── server.py          # Server instance
├── main.py            # Entry point
├── pyproject.toml     # Dependencies
└── README.md          # Documentation

模块化服务器设置

服务器.py -定义共享服务器实例：

from mcp.server.fastmcp import FastMCP

# Shared MCP server instance
mcp = FastMCP("my_server")

main.py -运行服务器的入口点：

from server import mcp

if __name__ == "__main__":
    mcp.run()

创建工具

工具是MCP服务器的核心功能。以下是示例：

简单计算工具

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """Multiplies two numbers together."""
    return a * b

API集成工具

import requests

@mcp.tool()
def get_weather(city: str) -> dict:
    """Fetches current weather data for a given city."""
    api_key = os.getenv("WEATHER_API_KEY")
    url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
    
    response = requests.get(url)
    return response.json()

文件操作工具

@mcp.tool()
def read_file(filepath: str) -> str:
    """Reads and returns the content of a text file."""
    try:
        with open(filepath, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File '{filepath}' not found"

MCP中的资源

资源公开了模型可以访问的静态或动态数据：

@mcp.resource("config://settings")
def get_settings():
    """Provides configuration settings."""
    return {
        "version": "1.0",
        "features": ["tool1", "tool2"]
    }

与Claude Desktop集成

配置步骤

  1. 打开（或创建）Claude Desktop配置文件：

macOS/Linux：

code ~/Library/Application\ Support/Claude/claude_desktop_config.json

窗户：

code %APPDATA%\Claude\claude_desktop_config.json
  1. 添加您的服务器配置：
{
  "mcpServers": {
    "my_server": {
      "command": "uv",
      "args": [
        "--directory",
        "/ABSOLUTE/PATH/TO/your_project",
        "run",
        "main.py"
      ]
    }
  }
}

  1. 替换 /ABSOLUTE/PATH/TO/your_project 使用您的实际项目路径

  2. 重新启动克劳德桌面

  3. 在界面中查找工具图标（锤子）-您的工具现在可用！

运行服务器

地方发展

直接运行服务器：

python server.py

或者使用主入口点：

python main.py

测试工具

使用MCP CLI检查器测试您的服务器：

mcp dev server.py

这将打开一个交互式界面，用于：

  • 列出可用工具
  • 带有样本输入的测试工具
  • 调试服务器响应

最佳实践

  1. 清除工具名称：使用描述性函数名称来指示工具的功能
  2. 详细文档字符串：为AI编写全面的描述，以了解工具用途
  3. 类型提示：始终包含类型注释，以便正确生成模式
  4. 错误处理：实现try-catch块以实现强大的工具执行
  5. 日志记录：添加日志记录以跟踪服务器操作和调试问题
  6. 安全：切勿硬编码敏感数据；使用环境变量
  7. 验证：在处理之前验证输入以避免错误

示例：完整的天气服务器

这是一个结合了多个概念的完整示例：

import os
import logging
import requests
from fastmcp import FastMCP

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create server
mcp = FastMCP(name="Weather Server")

@mcp.tool()
def get_current_weather(city: str) -> dict:
    """Fetches current weather for the specified city."""
    logger.info(f"Fetching weather for {city}")
    
    api_key = os.getenv("WEATHER_API_KEY")
    if not api_key:
        return {"error": "API key not configured"}
    
    try:
        url = f"https://api.openweathermap.org/data/2.5/weather"
        params = {"q": city, "appid": api_key, "units": "metric"}
        
        response = requests.get(url, params=params)
        response.raise_for_status()
        
        data = response.json()
        return {
            "city": city,
            "temperature": data["main"]["temp"],
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }
    except Exception as e:
        logger.error(f"Error fetching weather: {e}")
        return {"error": str(e)}

@mcp.tool()
def compare_weather(city1: str, city2: str) -> str:
    """Compares weather between two cities."""
    weather1 = get_current_weather(city1)
    weather2 = get_current_weather(city2)
    
    if "error" in weather1 or "error" in weather2:
        return "Unable to compare - error fetching weather data"
    
    temp_diff = abs(weather1["temperature"] - weather2["temperature"])
    
    return f"{city1}: {weather1['temperature']}°C, {city2}: {weather2['temperature']}°C. Difference: {temp_diff}°C"

if __name__ == "__main__":
    mcp.run()

常见用例

MCP服务器非常适合：

  • 数据库查询：查询SQLite、PostgreSQL或其他数据库
  • API集成：连接到外部API（天气、股票、新闻）
  • 文件操作：读取、写入和管理文件
  • 电子邮件/短信：发送通知或消息
  • 云操作：部署或管理云资源
  • 代码执行：安全地运行代码片段
  • 数据处理：分析CSV、JSON或其他数据格式

高级主题

异步工具

对于I/O绑定操作，请使用异步工具：

import asyncio

@mcp.tool()
async def fetch_data_async(url: str) -> dict:
    """Asynchronously fetches data from a URL."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

自定义退货类型

返回复杂的数据结构：

from typing import List, Dict

@mcp.tool()
def analyze_data(numbers: List[float]) -> Dict[str, float]:
    """Analyzes a list of numbers and returns statistics."""
    return {
        "mean": sum(numbers) / len(numbers),
        "max": max(numbers),
        "min": min(numbers)
    }

错误处理模式

实施稳健的错误处理：

@mcp.tool()
def safe_divide(a: float, b: float) -> dict:
    """Safely divides two numbers with error handling."""
    try:
        if b == 0:
            return {"error": "Division by zero"}
        return {"result": a / b}
    except Exception as e:
        return {"error": f"Unexpected error: {str(e)}"}

部署

云部署选项

您可以将MCP服务器部署到各种平台：

  • 数字海洋：使用应用平台或Droplets
  • 亚马逊云服务：EC2、Lambda或ECS
  • Heroku：使用Git进行简单部署
  • 塞瓦拉：具有MCP模板的现代PaaS

示例：基本HTTP服务器

对于生产环境，将MCP服务器封装在HTTP端点中：

from fastapi import FastAPI
from fastmcp import FastMCP

app = FastAPI()
mcp = FastMCP("production_server")

@app.post("/mcp")
async def handle_mcp_request(request: dict):
    # Handle MCP protocol requests
    return mcp.handle_request(request)

故障排除

常见问题

服务器未出现在Claude Desktop中：

  • 检查配置文件路径是否正确
  • 验证项目的绝对路径
  • 确保 uv 在您的系统PATH中
  • 配置更改后重新启动Claude Desktop

工具未执行：

  • 检查终端中的错误日志
  • 验证工具名称和类型提示
  • 确保安装了所有依赖项
  • 使用MCP CLI检查器的测试工具

导入错误：

  • 激活您的虚拟环境
  • 重新安装软件包： pip install -r requirements.txt
  • 检查Python版本兼容性

资源

后续步骤

现在您已经了解了基础知识：

  1. 为您的特定用例构建自定义服务器
  2. 尝试不同的工具类型
  3. 与您喜爱的API集成
  4. 与MCP社区共享您的服务器
  5. 探索提示和资源等高级功能

贡献

欢迎通过以下方式为本教程做出贡献：

  • 添加更多示例
  • 改进文件
  • 共享您自己的MCP服务器
  • 报告问题或提出改进建议

许可证

本教程是开源的，可用于学习目的。

幸福大厦！ 🚀

对于问题或反馈，请随时打开问题或为此存储库做出贡献。

相关 MCP