✨解锁 AI Agent 新姿势！手把手教你用 Python 搭建 MCP 服务，对接沪深数据 API (保姆级教程)✨

标签: MCP, AI Agent, Python, API对接, 量化交易, 效率神器, 后端开发

作者: PGFA

哈喽 CSDN 的小伙伴们！👋 我是 PGFA！今天给大家分享一个超酷的技术——模型上下文协议 (Model Context Protocol, MCP)！是不是听起来有点高大上？🤔 别担心，跟着 PGFA 的脚步，小白也能轻松搞定！

你是不是也遇到过这样的场景？

  • 想让 AI 帮你分析最新的股票数据，结果只能手动复制粘贴 K 线图或者财务报表？😩
  • 想让 AI 帮你执行一些需要调用外部 API 的任务，却发现 AI 本身做不到？🤯
  • 看到大佬们都在玩 AI Agent (智能体)，自己也想动手试试，却不知道从何开始？

那这篇教程就是为你准备的！🥳 今天，PGFA 就带大家用 Python 打造一个 MCP 服务，对接智兔数服的沪深数据 API (https://www.zhituapi.com/hsstockapi.html)，让你的 AI Agent (比如 Cursor 或 Cline) 瞬间拥有访问实时股票数据的超能力！📈 告别繁琐操作，拥抱智能自动化，效率直接拉满！🔥

这篇文章你能学到：

  1. MCP 的基本工作原理 💡
  2. 如何用 Python 编写一个基础但完整的 MCP 服务 🐍
  3. 如何对接真实的外部 API (以智兔沪深数据为例) 🔗
  4. 如何在 Cursor 和 Cline 中配置并使用你的 MCP 服务 ✅
  5. 一些实用的小 Tips 和注意事项 📌

话不多说，上干货！👇

🛠️ 准备工作：磨刀不误砍柴工

在开始之前，请确保你拥有以下装备：

  1. Python 环境: 建议 Python 3.8 或更高版本。
  2. requests 库: 用于发起 HTTP 请求调用 API。如果没安装，命令行运行：pip install requests。
  3. 智兔数服账号和 Token: 你需要注册智兔数服 (https://www.zhituapi.com/) 并获取你的 API Token。非常重要！ 本教程代码中会使用 ZHITU_TOKEN_LIMIT_TEST 这个测试 Token，但它有调用次数限制且功能可能不全，实际使用请务必替换成你自己的有效 Token！
  4. MCP 客户端 (可选其一):

🧠 核心概念科普：MCP 是个啥？

简单来说，MCP 就是由 AI 公司 Anthropic (Claude 背后的公司) 提出的一个开放标准，它定义了一种让 AI 大模型能够安全、标准化地与外部工具或服务进行交互的方式。

想象一下：

  • 你的 AI (客户端，比如 Cursor) 想知道某只股票的最新价格。
  • 它不会直接去访问股票 API，而是通过标准输入 (stdin) 发送一个 JSON-RPC 格式的请求给你的 MCP 服务。
  • 你的 MCP 服务 (就是我们今天要写的 Python 脚本) 收到请求，知道 AI 想调用 “获取股票价格” 这个工具 (Tool)。
  • MCP 服务去调用真正的智兔 API，拿到股票价格。
  • MCP 服务把结果打包成 JSON-RPC 格式的响应，通过标准输出 (stdout) 还给 AI。
  • AI 收到结果，然后告诉你股票价格。

整个过程就像 AI 有了一个听话的、专门负责跑腿的小助手 (MCP 服务)，可以帮它连接外部世界！✨

小伙伴们如果还是有不懂的地方可以看博主之前有关MCP介绍的文章进行更加深入的了解🤯 AI Agent 开发必备技能！MCP 协议到底是个啥？告别胶水代码，让你的 AI 大模型直接“动手”！(PGFA 保姆级解析)

🚀 实战开始：一步步编写 MCP 服务代码

好了，理论讲完，撸起袖子加油干！💪 我们来创建 mcp_zhitou_server.py 文件。

Step 1: 导入库和基础配置

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import json
import requests
import logging
import traceback

# --- 配置 ---
# 日志配置，输出到标准错误流，避免干扰 stdout
# 这样调试信息就不会被 AI 客户端误认为是响应了
logging.basicConfig(level=logging.INFO, stream=sys.stderr, format='%(asctime)s - %(levelname)s - %(message)s')

# Zhitou API 配置
ZHITU_API_BASE_URL = "https://api.zhituapi.com/hs"
# 重要：请务必替换为你的有效 Token！测试 Token 有限制！
ZHITU_TOKEN = "ZHITU_TOKEN_LIMIT_TEST" # 暂时使用教程提供的测试 Token

logging.info("MCP 服务脚本初始化...")
if ZHITU_TOKEN == "ZHITU_TOKEN_LIMIT_TEST":
    logging.warning("⚠️ 警告：当前使用的是测试 Token (ZHITU_TOKEN_LIMIT_TEST)，功能和次数可能受限。请替换为您自己的有效 Token！")
  • 我们导入了必要的库。
  • 配置了日志，让调试信息打印到 stderr，这样 stdout 就能干净地输出 JSON-RPC 响应。
  • 定义了智兔 API 的基础 URL 和 Token (再次强调，一定要换成你自己的！)。
  • 加了个启动日志和 Token 警告，对新手更友好。

Step 2: 封装 API 调用函数

为了方便调用不同的智兔 API，我们写一个通用的请求函数：

# --- API 调用辅助函数 ---
def call_zhitou_api(endpoint_path, params=None):
    """
    调用 Zhitou API 的通用函数。

    Args:
        endpoint_path (str): API 的端点路径 (例如: "list/all", "capital/lrqs/000001")
        params (dict, optional): GET 请求的查询参数. Defaults to None.

    Returns:
        dict: 成功时返回 API 响应的 JSON 数据。
    Raises:
        ConnectionError: 如果 API 请求失败。
        ValueError: 如果 API 响应不是有效的 JSON。
    """
    url = f"{ZHITU_API_BASE_URL}/{endpoint_path}?token={ZHITU_TOKEN}"
    logging.info(f"准备调用 Zhitou API: {url}")
    try:
        # 设置超时，避免服务卡死
        response = requests.get(url, params=params, timeout=20)
        # 检查 HTTP 状态码，非 200 则抛异常
        response.raise_for_status()
        logging.info(f"API 调用成功，状态码: {response.status_code}")
        return response.json()
    except requests.exceptions.Timeout:
        logging.error(f"调用 Zhitou API 超时: {url}")
        raise TimeoutError(f"请求 Zhitou API 超时 ({url})")
    except requests.exceptions.RequestException as e:
        logging.error(f"调用 Zhitou API 失败: {e}")
        # 将底层错误包装成更通用的 ConnectionError
        raise ConnectionError(f"无法连接或请求 Zhitou API 失败: {e}")
    except json.JSONDecodeError as e:
         logging.error(f"解析 Zhitou API 响应 JSON 失败: {e}")
         raise ValueError(f"无法解析 Zhitou API 返回的 JSON 数据")
  • 这个函数负责拼接 URL、添加 Token、发送 GET 请求。
  • 增加了超时设置 (timeout=20)。
  • 使用了 response.raise_for_status() 来检查 HTTP 错误。
  • 添加了更详细的日志和更具体的异常类型，方便排查问题。

Step 3: 定义 MCP 工具函数

现在，我们要根据智兔的 API 文档，把每个想让 AI 使用的功能定义成一个 Python 函数。函数名将作为 MCP 的工具名。

# --- MCP 工具函数 ---
# 这些函数对应 MCP 客户端可以调用的工具
# 函数名建议清晰易懂，与 API 功能对应
# 使用 **kwargs 接收可能的多余参数，增加兼容性
# 每个函数最好有 docstring 描述其功能和所需参数

def get_stock_list(**kwargs):
    """获取基础 A 股股票列表 (代码, 名称, 交易所)。"""
    logging.info("执行工具: get_stock_list")
    return call_zhitou_api("list/all")

def get_new_stock_calendar(**kwargs):
    """获取新股日历 (申购信息, 上市日期等)。"""
    logging.info("执行工具: get_new_stock_calendar")
    return call_zhitou_api("list/new")

def get_company_profile(stock_code, **kwargs):
    """获取指定股票代码的上市公司简介。
    Args:
        stock_code (str): 股票代码, 例如 '000001'。
    """
    if not stock_code:
        raise ValueError("工具 'get_company_profile' 需要 'stock_code' 参数。")
    logging.info(f"执行工具: get_company_profile, stock_code={stock_code}")
    # 注意 endpoint_path 的拼接方式
    return call_zhitou_api(f"gs/gsjj/{stock_code}")

def get_capital_daily_trend(stock_code, **kwargs):
    """获取指定股票代码的每日资金流入趋势 (近十年)。
    Args:
        stock_code (str): 股票代码, 例如 '000001'。
    """
    if not stock_code:
        raise ValueError("工具 'get_capital_daily_trend' 需要 'stock_code' 参数。")
    logging.info(f"执行工具: get_capital_daily_trend, stock_code={stock_code}")
    return call_zhitou_api(f"capital/lrqs/{stock_code}")

def get_all_announcements(stock_code, **kwargs):
     """获取指定股票代码的历史所有公告列表。
     Args:
         stock_code (str): 股票代码, 例如 '000001'。
     """
     if not stock_code:
         raise ValueError("工具 'get_all_announcements' 需要 'stock_code' 参数。")
     logging.info(f"执行工具: get_all_announcements, stock_code={stock_code}")
     return call_zhitou_api(f"msg/sygg/{stock_code}")

# --- 可以在这里根据智兔 API 文档继续添加其他工具函数 ---
# 比如：获取风险警示股、指数树、根据分类找股票、获取涨跌停股池等等
# def get_st_stock_list(**kwargs): ...
# def get_index_tree(**kwargs): ...
# def get_stocks_by_category_code(category_code, **kwargs): ...
# def get_limit_up_pool(trade_date, **kwargs): ...
  • 每个函数对应一个 API 功能。
  • 函数名就是 AI 调用时使用的工具名 (例如 get_company_profile)。
  • 对于需要参数的 API (如股票代码 stock_code)，在函数签名中定义，并在函数内部进行校验。
  • 使用 docstring 描述函数功能和参数，这对于后续配置 MCP 客户端很有用。
  • 鼓励读者根据文档自行添加更多工具函数，提高服务的实用性！

Step 4: 创建工具映射

我们需要一个字典，把工具名和对应的 Python 函数关联起来：

# --- MCP 工具映射 ---
# 将 MCP 客户端请求的工具名称映射到上面的 Python 函数
# 键名必须与客户端配置或 AI 调用时使用的工具名完全一致
TOOLS = {
    "get_stock_list": get_stock_list,
    "get_new_stock_calendar": get_new_stock_calendar,
    "get_company_profile": get_company_profile,
    "get_capital_daily_trend": get_capital_daily_trend,
    "get_all_announcements": get_all_announcements,
    # --- 如果你在上面添加了更多工具函数，记得在这里添加映射 ---
    # "get_st_stock_list": get_st_stock_list,
    # "get_index_tree": get_index_tree,
    # "get_stocks_by_category_code": get_stocks_by_category_code,
    # "get_limit_up_pool": get_limit_up_pool,
}
logging.info(f"已注册的 MCP 工具: {list(TOOLS.keys())}")

Step 5: 编写请求处理逻辑

这是 MCP 服务的核心，负责解析来自客户端的 JSON-RPC 请求，并调用相应的工具函数：

# --- 请求处理函数 ---
def handle_request(request_data):
    """处理来自 MCP 客户端的单个 JSON-RPC 请求"""
    request_id = request_data.get("id") # 获取请求 ID，响应时需要原样返回

    # 基础 JSON-RPC 格式校验
    if request_data.get("jsonrpc") != "2.0" or "method" not in request_data:
        logging.error(f"无效的请求格式 (非 JSON-RPC 2.0 或缺少 method): {request_data}")
        return json.dumps({
            "jsonrpc": "2.0",
            "error": {"code": -32600, "message": "无效请求 (Invalid Request)"},
            "id": request_id
        })

    method = request_data.get("method")
    params = request_data.get("params", {})

    response_payload = None # 初始化响应载荷

    # 处理 'tools/call' 方法 (AI 调用工具)
    if method == "tools/call":
        tool_name = params.get("name")
        # MCP 参数通常在 'arguments' 字段下，是一个字典
        arguments = params.get("arguments", {})
        logging.info(f"收到工具调用请求 -> Tool: '{tool_name}', Arguments: {arguments}")

        if tool_name in TOOLS:
            try:
                tool_function = TOOLS[tool_name]
                # 使用 **arguments 将参数字典解包为关键字参数传入函数
                result_data = tool_function(**arguments)
                # MCP 规定 result 必须是字符串，所以我们将 API 返回的 dict/list 转为 JSON 字符串
                response_payload = {
                    "result": json.dumps(result_data, ensure_ascii=False), # ensure_ascii=False 保留中文
                }
                logging.info(f"工具 '{tool_name}' 执行成功。")
            except (ValueError, TypeError) as e: # 参数错误
                 logging.warning(f"工具 '{tool_name}' 参数错误: {e}")
                 response_payload = {
                     "error": {"code": -32602, "message": f"无效参数 (Invalid params): {e}"}
                 }
            except (ConnectionError, TimeoutError, ValueError) as e: # API 调用或解析错误
                 logging.error(f"调用 Zhitou API 时出错 (工具 '{tool_name}'): {e}")
                 response_payload = {
                     "error": {"code": -32000, "message": f"服务器错误: 调用外部 API 失败 - {e}"}
                 }
            except Exception as e: # 其他未知错误
                logging.error(f"执行工具 '{tool_name}' 时发生未知错误: {e}\n{traceback.format_exc()}")
                response_payload = {
                     "error": {"code": -32000, "message": f"服务器内部错误: {type(e).__name__}: {e}"}
                 }
        else:
            logging.warning(f"请求的工具 '{tool_name}' 未在本服务中定义。")
            response_payload = {
                "error": {"code": -32601, "message": f"方法未找到 (Method not found): 工具 '{tool_name}' 不存在"}
            }

    # 处理 'tools/list' 方法 (客户端查询可用工具列表)
    elif method == "tools/list":
         logging.info("收到工具列表请求 (tools/list)")
         tool_list = []
         for name, func in TOOLS.items():
             # 尝试提取函数签名中的参数信息 (这是一个简化版本)
             # 更完善的方案可以使用 inspect 模块
             # 这里我们简单地从 docstring 中提取信息
             doc = func.__doc__ or "无描述"
             param_info = {}
             # 简单的基于 docstring 的参数提取 (可能不完全准确)
             if "Args:" in doc:
                 try:
                     args_section = doc.split("Args:")[1].split("Returns:")[0].split("Raises:")[0].strip()
                     for line in args_section.split('\n'):
                         parts = line.strip().split('(')
                         if len(parts) > 1:
                             param_name = parts[0].strip()
                             # 简单类型推断 (更复杂的需要 inspect)
                             param_type = "string" # 默认为 string
                             if "int" in parts[1].lower() or "number" in parts[1].lower():
                                 param_type = "number"
                             elif "bool" in parts[1].lower():
                                  param_type = "boolean"
                             param_info[param_name] = {"type": param_type}
                 except Exception as e:
                     logging.warning(f"解析工具 '{name}' 的 docstring 参数失败: {e}")

             tool_list.append({
                 "name": name,
                 "description": doc.split("Args:")[0].strip(), # 只取描述部分
                 "parameters": param_info # 添加参数信息
                 })
         # 同样，结果需要是 JSON 字符串
         response_payload = {"result": json.dumps(tool_list, ensure_ascii=False)}
         logging.info(f"返回工具列表: {tool_list}")

    # 不支持的方法
    else:
        logging.warning(f"收到不支持的方法请求: {method}")
        response_payload = {
            "error": {"code": -32601, "message": f"方法未找到 (Method not found): {method}"}
        }

    # 构建最终的 JSON-RPC 响应
    final_response = {
        "jsonrpc": "2.0",
        "id": request_id, # 必须包含原始请求 ID
    }
    # 合并 result 或 error
    final_response.update(response_payload)

    return json.dumps(final_response, ensure_ascii=False) # 确保中文字符正确编码
  • 严格按照 JSON-RPC 2.0 规范处理请求和响应。
  • 区分了 tools/call (调用工具) 和 tools/list (获取工具列表) 两种方法。
  • tools/call 会根据 name 查找 TOOLS 字典，找到对应的函数，并将 arguments 解包后传递给函数。
  • tools/list 会遍历 TOOLS 字典，提取函数名和 docstring 作为描述，返回给客户端。增加了简单的参数信息提取。
  • 对可能发生的错误（参数错误、API 调用失败、未知错误）进行了捕获和处理，返回符合规范的 error 对象。
  • 注意: result 字段的值必须是字符串，所以我们用 json.dumps() 将 API 返回的 Python 对象（字典或列表）序列化。ensure_ascii=False 保证中文字符不会被转义。

Step 6: 编写主循环

最后，我们需要一个主循环来持续监听 stdin，读取请求，调用处理函数，并将响应写回 stdout。

# --- 主循环 ---
def main():
    """MCP 服务器主入口点，持续监听 stdin"""
    logging.info("✅ Zhitou API MCP 服务已启动，开始监听标准输入 (stdin)...")
    while True:
        try:
            # 持续读取标准输入流的每一行
            line = sys.stdin.readline()
            # 如果读到文件末尾 (EOF)，说明客户端断开连接，退出循环
            if not line:
                logging.info("标准输入 (stdin) 已关闭，MCP 服务退出。")
                break

            line = line.strip()
            # 忽略空行
            if not line:
                continue

            logging.debug(f"收到原始输入行: {line}")

            try:
                # 解析收到的 JSON 请求
                request_data = json.loads(line)
            except json.JSONDecodeError:
                # 如果输入不是有效的 JSON，记录错误并跳过，不发送响应
                logging.error(f"❌ 无法解析输入的 JSON 数据: {line}")
                # 不返回错误响应，避免客户端因错误格式导致死循环
                continue

            # 处理解析后的请求数据
            response_str = handle_request(request_data)

            # 如果有响应字符串，则打印到标准输出
            if response_str:
                logging.debug(f"准备发送响应: {response_str}")
                # 打印响应到 stdout，flush=True 确保立即发送，不缓冲
                print(response_str, flush=True)

        except KeyboardInterrupt:
            # 捕获 Ctrl+C 信号，优雅退出
            logging.info("收到 KeyboardInterrupt (Ctrl+C)，正在关闭 MCP 服务...")
            break
        except BrokenPipeError:
             # 当客户端意外关闭连接时可能会发生
             logging.warning("管道已断开 (BrokenPipeError)，可能是客户端已关闭。MCP 服务退出。")
             break
        except Exception as e:
            # 捕获主循环中其他未预料到的异常
            logging.error(f"❌ MCP 服务主循环发生严重错误: {e}\n{traceback.format_exc()}")
            # 遇到严重错误时，可以选择退出或尝试继续
            # 这里选择记录错误后继续，尝试处理下一个请求
            continue

    logging.info("👋 MCP 服务已停止。")

# 脚本执行入口
if __name__ == "__main__":
    main()
  • 使用 while True 无限循环，持续监听。
  • sys.stdin.readline() 读取一行输入。如果读到空（EOF），表示客户端断开，退出循环。
  • flush=True 在 print() 中非常重要，确保响应立刻发送给客户端，而不是被 Python 的输出缓冲区缓存。
  • 增加了对 KeyboardInterrupt (Ctrl+C) 和 BrokenPipeError 的处理，让服务退出更优雅。

🎉 激动人心的完整代码

好了，把上面的步骤整合起来，就是我们完整的 mcp_zhitou_server.py 啦！

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import json
import requests
import logging
import traceback

# --- 配置 ---
logging.basicConfig(level=logging.INFO, stream=sys.stderr, format='%(asctime)s - %(levelname)s - %(message)s')

ZHITU_API_BASE_URL = "https://api.zhituapi.com/hs"
# 🔥🔥🔥【请务必替换为你自己的有效 Token! 测试 Token 有限制!】🔥🔥🔥
ZHITU_TOKEN = "ZHITU_TOKEN_LIMIT_TEST"

logging.info("MCP 服务脚本初始化...")
if ZHITU_TOKEN == "ZHITU_TOKEN_LIMIT_TEST":
    logging.warning("⚠️ 警告：当前使用的是测试 Token (ZHITU_TOKEN_LIMIT_TEST)，功能和次数可能受限。请替换为您自己的有效 Token！")


# --- API 调用辅助函数 ---
def call_zhitou_api(endpoint_path, params=None):
    url = f"{ZHITU_API_BASE_URL}/{endpoint_path}?token={ZHITU_TOKEN}"
    logging.info(f"准备调用 Zhitou API: {url}")
    try:
        response = requests.get(url, params=params, timeout=20)
        response.raise_for_status()
        logging.info(f"API 调用成功，状态码: {response.status_code}")
        # 尝试解析 JSON
        try:
            data = response.json()
            # 可以在这里检查智兔 API 可能返回的业务错误码（如果文档有定义）
            # 例如: if data.get('code') != 0: logging.error(...) raise ValueError(...)
            return data
        except json.JSONDecodeError as e:
             logging.error(f"解析 Zhitou API 响应 JSON 失败: {e}. 响应内容: {response.text[:500]}...") # 记录部分响应内容
             raise ValueError(f"无法解析 Zhitou API 返回的 JSON 数据")
    except requests.exceptions.Timeout:
        logging.error(f"调用 Zhitou API 超时: {url}")
        raise TimeoutError(f"请求 Zhitou API 超时 ({url})")
    except requests.exceptions.RequestException as e:
        logging.error(f"调用 Zhitou API 失败: {e}")
        raise ConnectionError(f"无法连接或请求 Zhitou API 失败: {e}")


# --- MCP 工具函数 ---
def get_stock_list(**kwargs):
    """获取基础 A 股股票列表 (代码, 名称, 交易所)。"""
    logging.info("执行工具: get_stock_list")
    return call_zhitou_api("list/all")

def get_new_stock_calendar(**kwargs):
    """获取新股日历 (申购信息, 上市日期等)。"""
    logging.info("执行工具: get_new_stock_calendar")
    return call_zhitou_api("list/new")

def get_company_profile(stock_code, **kwargs):
    """获取指定股票代码的上市公司简介。
    Args:
        stock_code (str): 股票代码, 例如 '000001'。
    """
    if not stock_code:
        raise ValueError("工具 'get_company_profile' 需要 'stock_code' 参数。")
    logging.info(f"执行工具: get_company_profile, stock_code={stock_code}")
    return call_zhitou_api(f"gs/gsjj/{stock_code}")

def get_capital_daily_trend(stock_code, **kwargs):
    """获取指定股票代码的每日资金流入趋势 (近十年)。
    Args:
        stock_code (str): 股票代码, 例如 '000001'。
    """
    if not stock_code:
        raise ValueError("工具 'get_capital_daily_trend' 需要 'stock_code' 参数。")
    logging.info(f"执行工具: get_capital_daily_trend, stock_code={stock_code}")
    return call_zhitou_api(f"capital/lrqs/{stock_code}")

def get_all_announcements(stock_code, **kwargs):
     """获取指定股票代码的历史所有公告列表。
     Args:
         stock_code (str): 股票代码, 例如 '000001'。
     """
     if not stock_code:
         raise ValueError("工具 'get_all_announcements' 需要 'stock_code' 参数。")
     logging.info(f"执行工具: get_all_announcements, stock_code={stock_code}")
     return call_zhitou_api(f"msg/sygg/{stock_code}")

# --- 你可以继续添加更多工具函数在这里 ---


# --- MCP 工具映射 ---
TOOLS = {
    "get_stock_list": get_stock_list,
    "get_new_stock_calendar": get_new_stock_calendar,
    "get_company_profile": get_company_profile,
    "get_capital_daily_trend": get_capital_daily_trend,
    "get_all_announcements": get_all_announcements,
    # --- 如果添加了新函数，在这里加上映射 ---
}
logging.info(f"已注册的 MCP 工具: {list(TOOLS.keys())}")


# --- 请求处理函数 ---
def handle_request(request_data):
    request_id = request_data.get("id")

    if request_data.get("jsonrpc") != "2.0" or "method" not in request_data:
        logging.error(f"无效的请求格式: {request_data}")
        return json.dumps({
            "jsonrpc": "2.0",
            "error": {"code": -32600, "message": "无效请求 (Invalid Request)"},
            "id": request_id
        })

    method = request_data.get("method")
    params = request_data.get("params", {})
    response_payload = None

    if method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        logging.info(f"收到工具调用请求 -> Tool: '{tool_name}', Arguments: {arguments}")

        if tool_name in TOOLS:
            try:
                tool_function = TOOLS[tool_name]
                result_data = tool_function(**arguments)
                response_payload = {"result": json.dumps(result_data, ensure_ascii=False)}
                logging.info(f"工具 '{tool_name}' 执行成功。")
            except (ValueError, TypeError) as e:
                 logging.warning(f"工具 '{tool_name}' 参数错误: {e}")
                 response_payload = {"error": {"code": -32602, "message": f"无效参数 (Invalid params): {e}"}}
            except (ConnectionError, TimeoutError, ValueError) as e:
                 logging.error(f"调用 Zhitou API 时出错 (工具 '{tool_name}'): {e}")
                 response_payload = {"error": {"code": -32000, "message": f"服务器错误: 调用外部 API 失败 - {e}"}}
            except Exception as e:
                logging.error(f"执行工具 '{tool_name}' 时发生未知错误: {e}\n{traceback.format_exc()}")
                response_payload = {"error": {"code": -32000, "message": f"服务器内部错误: {type(e).__name__}: {e}"}}
        else:
            logging.warning(f"请求的工具 '{tool_name}' 未在本服务中定义。")
            response_payload = {"error": {"code": -32601, "message": f"方法未找到 (Method not found): 工具 '{tool_name}' 不存在"}}

    elif method == "tools/list":
         logging.info("收到工具列表请求 (tools/list)")
         tool_list = []
         for name, func in TOOLS.items():
             doc = func.__doc__ or "无描述"
             param_info = {}
             if "Args:" in doc:
                 try:
                     args_section = doc.split("Args:")[1].split("Returns:")[0].split("Raises:")[0].strip()
                     for line in args_section.split('\n'):
                         parts = line.strip().split('(')
                         if len(parts) > 1:
                             param_name = parts[0].strip()
                             param_type = "string"
                             if "int" in parts[1].lower() or "number" in parts[1].lower(): param_type = "number"
                             elif "bool" in parts[1].lower(): param_type = "boolean"
                             param_info[param_name] = {"type": param_type}
                 except Exception as e:
                     logging.warning(f"解析工具 '{name}' 的 docstring 参数失败: {e}")
             tool_list.append({
                 "name": name,
                 "description": doc.split("Args:")[0].strip(),
                 "parameters": param_info
                 })
         response_payload = {"result": json.dumps(tool_list, ensure_ascii=False)}
         logging.info(f"返回工具列表: {tool_list}")
    else:
        logging.warning(f"收到不支持的方法请求: {method}")
        response_payload = {"error": {"code": -32601, "message": f"方法未找到 (Method not found): {method}"}}

    final_response = {"jsonrpc": "2.0", "id": request_id}
    final_response.update(response_payload)
    return json.dumps(final_response, ensure_ascii=False)


# --- 主循环 ---
def main():
    logging.info("✅ Zhitou API MCP 服务已启动，开始监听标准输入 (stdin)...")
    while True:
        try:
            line = sys.stdin.readline()
            if not line:
                logging.info("标准输入 (stdin) 已关闭，MCP 服务退出。")
                break
            line = line.strip()
            if not line: continue
            logging.debug(f"收到原始输入行: {line}")
            try:
                request_data = json.loads(line)
            except json.JSONDecodeError:
                logging.error(f"❌ 无法解析输入的 JSON 数据: {line}")
                continue
            response_str = handle_request(request_data)
            if response_str:
                logging.debug(f"准备发送响应: {response_str}")
                print(response_str, flush=True) # 关键！确保立即发送！
        except KeyboardInterrupt:
            logging.info("收到 KeyboardInterrupt (Ctrl+C)，正在关闭 MCP 服务...")
            break
        except BrokenPipeError:
             logging.warning("管道已断开 (BrokenPipeError)，可能是客户端已关闭。MCP 服务退出。")
             break
        except Exception as e:
            logging.error(f"❌ MCP 服务主循环发生严重错误: {e}\n{traceback.format_exc()}")
            continue
    logging.info("👋 MCP 服务已停止。")

# 脚本执行入口
if __name__ == "__main__":
    main()

⚙️ 配置与使用：让 AI 用起来！

代码写好了，怎么让 Cursor 或 Cline 用起来呢？很简单！

  1. 保存文件: 将上面的完整代码保存为 mcp_zhitou_server.py。

  2. 安装依赖: 再次确认 requests 已安装 (pip install requests)。

  3. 🔥【极其重要】替换 Token🔥: 打开 mcp_zhitou_server.py，找到 ZHITU_TOKEN = "ZHITU_TOKEN_LIMIT_TEST" 这一行，把测试 Token 换成你自己的智兔数服 API Token！ 否则 API 调用会失败或受限！

  4. 配置 Cline:

    • 打开 Cline (无论是独立版还是 VS Code 插件)。
    • 找到管理 MCP Servers 的地方 (通常在设置里)。
    • 点击添加/新建 MCP Server。
    • Name (名称): ZhitouAPI (或者你喜欢的名字)
    • Command (命令): 填入执行 Python 脚本的命令，注意使用绝对路径或确保脚本在 PATH 中。
      • 示例 (Mac/Linux): python3 /Users/pgfa/scripts/mcp_zhitou_server.py
      • 示例 (Windows): python C:\Users\PGFA\Documents\mcp_zhitou_server.py
      • 注意: PGFA是博主自己电脑上的用户名这里需要替换成你们电脑上的路径哦
    • 保存。Cline 会自动启动这个服务进程。你应该能在 Cline 的 MCP 管理界面看到它显示为“已连接”或类似的绿色状态。

  5. 配置 Cursor:

    • 在你的项目根目录创建一个 .cursor 文件夹。
    • 在 .cursor 文件夹里创建一个 mcp.json 文件。
    • 将以下内容粘贴到 mcp.json，并修改 args 中的脚本路径为你的实际路径：
    {
  "mcpServers": {
    "zhitou": {
      "command": "python", // 或 python3，取决于你的环境
      "args": ["/Users/pgfa/scripts/mcp_zhitou_server.py"], // ⚠️ 修改为你的脚本实际路径！
      "workingDirectory": "/Users/pgfa/scripts/", // ⚠️ 可选，建议设为脚本所在目录
      "tools": [
        // 这里最好把你实现的工具都声明一下，帮助 Agent 理解
        {"name": "get_stock_list", "description": "获取基础 A 股股票列表 (代码, 名称, 交易所)。"},
        {"name": "get_new_stock_calendar", "description": "获取新股日历 (申购信息, 上市日期等)。"},
        {"name": "get_company_profile", "description": "获取指定股票代码的上市公司简介。", "parameters": {"stock_code": "string"}},
        {"name": "get_capital_daily_trend", "description": "获取指定股票代码的每日资金流入趋势 (近十年)。", "parameters": {"stock_code": "string"}},
        {"name": "get_all_announcements", "description": "获取指定股票代码的历史所有公告列表。", "parameters": {"stock_code": "string"}}
        // ... 其他你添加的工具 ...
      ]
    }
  }
}
    • 保存文件。重启 Cursor 或重新加载项目后，Cursor Agent 应该就能识别并使用这个 MCP 服务了。你可以在 Cursor 的设置 -> MCP 部分查看状态。

  6. 开始提问！
    现在，你可以在 Cline 或 Cursor 的聊天窗口里，像和真人对话一样，让 AI 调用你的 Zhitou API 工具啦！试试看：

    • “帮我查一下 000001 平安银行的公司简介。”
    • “获取最新的 A 股列表。”
    • “看看 600519 贵州茅台最近的资金流入情况。” (如果实现了 get_capital_daily_trend)
    • “今天的新股日历有哪些？”

    观察 AI 的回复，以及你运行 MCP 服务脚本的终端（或 Cline/Cursor 的日志），看看请求是否被正确处理，API 是否被成功调用！是不是超级方便？！🤩

💡 进阶思考与小贴士

  • Token 安全: 生产环境中，千万不要把 Token 硬编码！最好使用环境变量 (os.environ.get('ZHITU_API_TOKEN')) 或者专门的配置文件/密钥管理服务。
  • 添加更多工具: 这个框架很容易扩展，只要对着智兔 API 文档，添加新的 Python 函数和 TOOLS 映射就行。让你的 AI 助手更全能！
  • 更精细的错误处理: 可以根据 Zhitou API 可能返回的特定错误码，给出更友好的错误提示。
  • 异步处理: 如果你觉得API 调用比较耗时，或者你想处理并发请求，可以考虑使用 asyncio 将服务改造成异步的。
  • Windows 路径: Windows 用户请注意路径分隔符是 \，在 Python 字符串中可能需要写成 \\ 或使用原始字符串 r"C:\path"。

总结

好啦！今天 PGFA 带大家从 0 到 1 搭建了一个实用的 Python MCP 服务，成功让 AI Agent 对接了沪深数据 API。是不是感觉 AI 的能力边界又被拓宽了？🤯 通过 MCP，我们可以赋予 AI 调用各种外部工具的能力，无论是查询数据库、控制智能家居，还是像今天这样获取金融数据，都变得触手可及！

希望这篇保姆级教程对你有帮助！如果你觉得有用，别忘了点赞👍、收藏🌟、关注 PGFA 哦！ 也欢迎在评论区留言交流你的想法和遇到的问题，我们一起探讨学习！下次再给大家带来更多好玩的技术分享！拜拜~ ✨

# 人工智能 # python # 开发语言
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

【MCP】从0到1实现一个MCP Client

青蛇》01 MCP回顾MCP全称Model Context Protocol，模型上下文协议。上一篇文章【MCP】从0到1实现一个MCP Server 我们介绍了如何从0~1实现一个自定义的 MCP Server，感兴趣的朋友可以点击文章链接回顾下或者查看下。02 MCP Client本质。

avatar MCP技术社区
cover

RAG 效果不佳？试试 MCP+数据库（Text2SQL）：提升大模型结构化数据检索精度的实战指南

avatar MCP技术社区
cover

让 AI 更懂 Ant Design：MCP 协议在前端领域的落地实践，收藏这一篇就够了！！

avatar MCP技术社区