手搓第一个 MCP Server：实现水质在线数据查询

随着人工智能技术的快速发展，如何将大语言模型（LLM）与实际业务场景结合，提供精准、可控的服务成为一个热门话题。MCP（Model Context Protocol）作为一种开放协议，为应用程序向 LLM 提供标准化的上下文接口，极大地简化了这一过程。本文将以构建一个水质在线查询 MCP 服务为例，手把手带你从零开始实现一个功能强大且实用的 MCP Server。通过这个过程，你将深入理解 MCP 的核心机制，并掌握如何将其应用于实际场景。

什么是 MCP？

MCP（Model Context Protocol）是一个为人工智能应用程序设计的标准化协议，旨在为大语言模型（LLM）提供统一的上下文输入和工具调用接口。可以将 MCP 比喻为 AI 世界的“USB-C 接口”：正如 USB-C 为设备连接外设提供了通用标准，MCP 为 AI 模型连接数据源、工具和业务逻辑提供了规范化的桥梁。

通过 MCP，开发者可以定义结构化的工具（Tools），让 LLM 根据用户需求智能调用这些工具，获取实时数据或执行特定任务。相比直接依赖 LLM 的生成能力，MCP 能够显著减少模型“幻觉”（生成不准确或虚构内容），提供更可靠的结果。

MCP 的核心优势

标准化：提供统一的输入输出规范，兼容多种 LLM 和工具。
灵活性：支持从本地数据查询到远程 API 调用的各种场景。
可控性：通过工具调用，开发者可以精确控制模型的行为和输出。
可扩展性：便于将服务部署到本地或云端，供个人或团队使用。

官方文档：MCP 简介 - MCP 中文文档

从天气查询示例学习 MCP

为了更好地理解 MCP 的工作原理，我们先来看一个简单的天气查询 MCP 示例。这个示例通过调用美国国家气象服务（NWS）的 API，实现了天气预报和警报查询功能。以下是核心代码和逻辑分析：

示例代码：天气查询 MCP

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 初始化 FastMCP 服务
mcp = FastMCP("weather")

# 常量定义
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

# 辅助函数：发送请求并格式化数据
async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向 NWS API 发送请求，并进行错误处理。"""
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

def format_alert(feature: dict) -> str:
    """将警报数据格式化为可读字符串。"""
    props = feature["properties"]
    return f"""
事件: {props.get('event', 'Unknown')}
区域: {props.get('areaDesc', 'Unknown')}
严重性: {props.get('severity', 'Unknown')}
描述: {props.get('description', 'No description available')}
指示: {props.get('instruction', 'No specific instructions provided')}
"""

# 工具定义
@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取美国州的天气警报。"""
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)
    if not data or "features" not in data:
        return "无法获取警报或未找到警报。"
    if not data["features"]:
        return "该州没有活跃的警报。"
    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取指定位置的天气预报。"""
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)
    if not points_data:
        return "无法获取此位置的预报数据。"
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data:
        return "无法获取详细预报。"
    periods = forecast_data["properties"]["periods"]
    forecasts = [f"""
{period['name']}:
温度: {period['temperature']}°{period['temperatureUnit']}
风: {period['windSpeed']} {period['windDirection']}
预报: {period['detailedForecast']}
""" for period in periods[:5]]
    return "\n---\n".join(forecasts)

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

代码解析

初始化 MCP 服务：通过 FastMCP("weather") 创建一个名为“weather”的 MCP 服务。
定义辅助函数：
- make_nws_request：请求 NWS API。
- format_alert：将 API 返回的警报数据格式化为用户友好的字符串。
定义工具：
- get_alerts：根据州代码查询天气警报。
- get_forecast：根据经纬度查询天气预报。
运行服务：通过 mcp.run(transport="stdio") 启动服务，等待 LLM 调用。

这个示例展示了 MCP 的核心思想：通过标准化的工具接口，将外部数据（天气 API）与 LLM 的自然语言处理能力结合，提供精准的查询结果。

动手实践：构建水质查询 MCP

基于天气查询的经验，我们现在来实现一个水质在线查询 MCP 服务。该服务允许用户通过自然语言输入时间范围、站点名称和监测因子，查询水质监测数据。以下是完整实现步骤和代码。

开发环境准备

初始化 Python 项目：
使用 uv 工具初始化项目并安装依赖：
1
2
3
uv init firstmcp cd firstmcp uv add mcp[cli] httpx pandas pymysql
数据源说明：
水质数据可以来自多种来源，例如：
- 远程 API 接口
- MySQL 数据库
- Excel 文件
  本示例假设数据通过 multi_station_data 函数从数据库或文件中获取，返回一个 Pandas DataFrame。

核心代码：水质查询 MCP

import pandas as pd
from datetime import datetime
from typing import Any
from mcp.server.fastmcp import FastMCP

# 初始化 MCP 服务
mcp = FastMCP("water_quality")

def multi_station_data(stations: list, start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """
    获取多个站点在指定时间范围内的水质监测数据。

    Args:
        stations: 站点名称列表
        start_date: 开始时间
        end_date: 结束时间

    Returns:
        DataFrame: 包含监测数据的表格
    """
    # 示例：模拟从数据库或 API 获取数据
    # 实际实现中可替换为 MySQL 查询、API 调用或 Excel 读取
    # 以下仅为示例
    data = {
        "监测时间": [datetime(2025, 5, 1, 10, 0), datetime(2025, 5, 1, 11, 0)],
        "设备名称": [stations[0], stations[0]],
        "氨氮": [0.5, 0.6],
        "总磷": [0.2, 0.25],
        "高锰酸盐指数": [2.1, 2.3]
    }
    return pd.DataFrame(data)

@mcp.tool()
async def get_water_quality(
    start_time: str,
    end_time: str,
    station_name: str,
    factors: str = None
) -> str:
    """
    查询指定站点和时间范围内的水质监测数据。

    Args:
        start_time: 开始时间 (格式: YYYY-MM-DD HH:MM:SS)
        end_time: 结束时间 (格式: YYYY-MM-DD HH:MM:SS)
        station_name: 站点名称
        factors: 逗号分隔的因子名称 (例如: '氨氮,总磷,高锰酸盐指数')，可选

    Returns:
        str: 格式化的查询结果
    """
    try:
        # 解析时间参数
        start_date = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
        end_date = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
        
        # 查询数据
        df_water = multi_station_data([station_name], start_date, end_date)
        if df_water is None or df_water.empty:
            return "未找到符合条件的监测数据。"
        
        # 确保时间格式正确
        df_water["监测时间"] = pd.to_datetime(df_water["监测时间"])
        
        # 筛选指定因子
        if factors:
            factors_list = [factor.strip() for factor in factors.split(",")]
            valid_factors = [f for f in factors_list if f in df_water.columns]
            if not valid_factors:
                return "指定的因子在监测数据中不存在。"
            df_water = df_water[["监测时间", "设备名称"] + valid_factors]
        
        # 格式化输出
        result = df_water.to_string(index=False)
        return result
    
    except ValueError as e:
        return f"参数错误: {e}"
    except Exception as e:
        return f"发生错误: {e}"

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

代码解析

初始化 MCP 服务：
- 使用 FastMCP("water_quality") 创建一个名为“water_quality”的服务。
数据获取函数：
- multi_station_data 是一个占位函数，模拟从数据源获取水质数据。实际应用中，你可以替换为：
  - MySQL 查询：使用 pymysql 连接数据库。
  - API 调用：使用 requests 或 httpx 获取远程数据。
  - Excel 读取：使用 pandas.read_excel。
工具定义：
- get_water_quality 工具接受时间范围、站点名称和可选的因子列表，查询并返回格式化的水质数据。
- 使用 Pandas 处理数据，支持动态筛选因子。
错误处理：
- 捕获时间格式错误、数据缺失等异常，返回用户友好的提示。

集成到 Trae 平台

Trae 是一个支持 MCP 的 AI 平台（本文基于 Trae 1.3.3 版本），可以轻松将你的 MCP 服务集成到其生态中。以下是集成步骤：

打开 Trae 设置：
- 点击个人头像，选择“AI 功能管理” -> “MCP”。
- 点击“添加”，选择“手动配置”。

配置 MCP 服务：

复制以下 JSON 配置到配置框，注意将修改为你项目的实际路径：

{
  "mcpServers": {
    "waterquality": {
      "command": "uv",
      "args": [
        "--directory",
        "D:/python/learn/first_mcp/firstmcp",
        "run",
        "single.py"
      ]
    }
  }
}

确认并测试：
- 点击“确认”，Trae 会验证配置。如果成功，会显示 ✅ 并提示服务可用。
- 在 Trae 对话界面输入类似“查询 2025-05-01 10:00:00 至 2025-05-01 12:00:00 的某站点氨氮数据”，即可测试效果。

测试效果

以下是一个测试示例（假设使用 Claude 3.5 模型）：

用户输入：查询某站点最新的高指数据数据。
模型输出：成功分析出所需要的入参和查询结果，并自动判断是否超标。

测试效果截图

部署到服务器（进阶）

目前，我们的服务运行在本地。如果希望团队或公众使用，可以将 MCP 服务部署到服务器，公开部署后，其他用户可以像添加 Cherry Studio 的公开 MCP 服务一样，轻松接入你的水质查询服务。

MCP 的应用价值与思考

通过以上实践，我们成功构建了一个水质查询 MCP 服务。这只是 MCP 应用的冰山一角。MCP 的真正价值在于其标准化的工具调用机制，能够将 LLM 的语言理解能力与结构化数据和业务逻辑无缝结合。

总结

本文通过一个水质查询 MCP 服务的实现，详细介绍了 MCP 的核心概念、开发流程和集成方法。从天气查询示例到水质查询实践，我们看到 MCP 如何将 LLM 与外部数据源结合，提供精准、可控的服务。通过集成到 Trae 平台和可能的服务器部署，MCP 展现了从本地开发到广泛应用的潜力。

MCP 不仅是一种技术工具，更是一种设计智能体的思维方式。它鼓励开发者将业务逻辑封装为标准化的工具，让 AI 更好地服务于现实世界。希望本文能启发你动手实践，探索 MCP 在更多场景中的可能性！

参考资料

人工智能

#python #AIGC #mcp

手搓第一个 MCP Server：实现水质在线数据查询

https://maoyu92.github.io/2025/05/02/06 人工智能/手搓一个 MCP Server：实现水质在线数据查询/

作者

陈文茂

发布于

2025年5月2日

许可协议

AI打造个性化审稿专家上一篇

UV入门指南：更快的Python包管理工具实践下一篇