大模型接入自定义 MCP Server，我制作了个免费使用的基金涨跌归纳和归因分析的 Agent

月小水长

发布于 2025-11-12 20:08:51

1810

文章被收录于专栏：月小水长月小水长

最近一年基金一路高歌猛进，已达十年之峰，

作为基民，不仅需要关注基金的涨跌详情，归纳分析，

还需要理解相关行业热点新闻对基金涨跌的影响，进行归因分析，

但是这需要太多时间了，在各个基金和新闻 App 来回切换，煞费时间，

于是，两个月前，我开发了一款利用大模型分析基金涨跌的 Web 工具，

核心思路是开发多个数据源 MCP Server，指定 Streamable HTTP 为传输方式，

然后使用 Gemini 大模型作为 Host，以 API 方式 Client Session 接入 MCP，

详情可见：

手搓了两个 MCP Server，独立开发基金涨跌大模型分析工具

用户只需要在前端界面，输入基金代码和选择时间段，就能得到大模型输出的精美分析报告，

当时我在文末说，部署该工具上线给大家免费使用（考虑到 Token 消耗...

一不留神就过了两个月了，于是上周日加班加点，兑现了这个承诺，而且升级工具成了 Agent，

注：在不同的领域 Agent 有不同的含义，本文限定讨论在大模型范畴下的 Agent 业界关于 Agent 的定义并没有明确，有人认为必须是有记忆和推理能力的 Workflow/Pipeline 才能叫 Agent，有的人则坚持只要有智能能力就能被称为 Agent，

这个分析基金归纳和归因的 Agent 使用效果如下：

在此特别强调声明，本工具只是一个 Funny Tool，不构成任何投资建议：

基金投资有风险，过往业绩不代表未来表现。 本文和本 Agent 生成内容仅为信息参考，不构成任何投资建议， 投资者据此操作产生的盈亏均需自行承担。

地址公开给大家，免费使用的，地址见文末，

下面我将介绍下开发的流程和心得，

主要包括自定义的两个 MCP Server 如何开发，如何部署上云，以及如何嵌入 Agent

基金涨跌数据 MCP Server

基金涨跌数据有很多 API，付费的免费的都有，

也有很多爬虫项目，包括模拟的接口的，

本文不关心数据的上游生产，只考虑基金数据作为消费者如何通过 MCP Server 暴露给大模型，

首先定义一个 MCP 服务实例，

from mcp.server.fastmcp import FastMCP

# 创建 MCP 实例
mcp = FastMCP(
    name="FundTrendQuery",
    instructions="一个基金涨跌查询器，可根据基金代码和时间段返回净值和涨跌详情。"
)

然后通过注解形式添加一个 MCP Tool，

除了 Tool，还有 Resource、Prompt、Image 等多种 annotation，感兴趣可以查看 MCP SDK

@mcp.tool()
async def query_fund_trend(
        fund: str,
        start_time: str | None = None,
        end_time: str | None = None,
) -> List[Dict[str, Any]]:
    """
    查询某基金在某个时间段的涨跌详情。

    参数：
    - fund: 基金代码，如 "110022"
    - start_time: 起始日期，格式 "YYYY-MM-DD"
    - end_time: 结束日期，格式 "YYYY-MM-DD"

    """
    
    # todo 接口实现

接口实现就是业务数据的获取实现，以对现有数据层 API 进行一层包装，最后在注解函数返回数据，

需要强调在 MCP Tool 中注解函数注释的重要性，无论是 Functionality 功能说明还是入口和出口的 Param 参数解释，

都是在向大模型根据上下文无缝调用该 Tool 的 Compatibility 提高更进一步，

最后需要注意，为了大模型的 Structured Output，Tool 函数返回的数据必须是 Pydantic、Primitive /Generic types 等类型

最后启动 MCP Server，

mcp.run(transport='sse')

MCP Server 主要有三种传输格式，这里使用 SSE transport

行业新闻数据 MCP Server

这个流程其实和基金涨跌数据差不多，照猫画虎就行，直接贴代码，不再赘述，

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="NewsQuery",
              instructions="一个新闻提取器，可根据关键词和时间段查询新闻数据。")

@mcp.tool()
async def query_news(
        keywords: list[str],
        start_time: str | None = None,
        end_time: str | None = None,
) -> List[Dict[str, Any]]:
    """
    根据 keywords 和时间范围查询新闻，返回新闻条目集合，keywords 可为空

    参数：
    - keywords: 关键词列表
    - start_time: 起始日期，格式 "YYYY-MM-DD"
    - end_time: 结束日期，格式 "YYYY-MM-DD"

    返回：JSON list[dict]
    """
    # todo 接口实现

# 启动 MCP Server
if __name__ == "__main__":
    mcp.run(transport='sse')

不过这里可能疑问了，怎么保证新闻数据实时性呢，当然是增量更新，

无论是基金还是新闻，我们可以选择在生产的时候增加数据持久层，然后定时更新，

不推荐在 MCP Server 里生产数据，可以使用另外的 Pipeline 定时更新，

不推荐选择在消费的时候发现没有才通知上游生产更新（消费体验会很差

上云部署可靠 MCP Server

和传统的 Flask/Django/FastApi 不太一样的是，MCP Server 本质就是一个 长连接 HTTP/WebSocket 服务

MCP 的 SDK 已经内置了 HTTP server (httpx/uvicorn-like) 去处理请求，

所以它不需要需要 WSGI/ASGI 容器（Gunicorn/uWSGI/Uvicorn），

但是为了可靠性，还是要部署一个守护进程，Systemd 或者 Circus 等服务均可，

为了充分利用 CPU 性能，进程数设置成核心数，

为了开启 SSL 证书和前端流量负载均衡，可以再用 Nginx 反向代理 MCP Server，

由于 MCP Server 的长连接 HTTP 本质，Nginx 的 Server 块中需要做一些特殊设置，

这也是本节的重点（因为本人踩过坑），特殊设置块如下：

proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s;
proxy_send_timeout 86400s;
keepalive_timeout 86400s;
send_timeout 86400s;

第一行 proxy_http_version 1.1;，意即把 Nginx Web 服务器与上游（upstream）应用服务器之间的 HTTP 协议版本设为 HTTP/1.1；

而 HTTP/1.1 支持长连接（keep-alive）**和**分块传输（chunked transfer encoding），这是 SSE Transport 的必要条件。

第二行 proxy_set_header Connection "";，指把转发给上游的 Connection 头设置为空（即不发送 Connection 头）；

可以避免 Nginx 自动添加 Connection: close 或其他值，这样可以保持 HTTP/1.1 的持久连接行为。

第三行和第四行效果类似，分别是关闭 Nginx 对上游响应的缓冲和关闭 Nginx 的代理缓存；道理很简单，流式或动态实时数据不应被缓存。

最后四行也是如此，SSE 连接通常需要保持很长时间，所以需要把各种 Timeout 超时拉长，避免 Nginx 在长时间没有数据流时主动断开，但是这就需要服务器承压了。