Model Context Protocol(MCP)的SSE使用

本文最后更新于 2025年4月11日 上午

通过一个简单的例子,展示如何使用 MCP 的SSE方式 快速构建服务。没有介绍和原理,单纯流程示例。

之前一直使用函数来调用接口实现各个功能,只要约束写接口的人文档规范,还是挺便利的,最近MCP大火,于是试试,结果发现,如果推行那么之前后端写接口这个工作变成了我来写MCP服务…好吧,说笑的.

找了半天没发现SSE怎么增加接口的权鉴,这么开放的么,还是说这协议主推stdio本地模式.完全打消切换的想法了.

本文是自己试过啥记录啥,体验过更多后会更新

安装与环境准备

一直使用的conda,没用过uv,并且反正又不是stdio模式本地运行,所有无所谓。
mcp只需要安装:

1
pip install mcp[cli]

创建一个简单的 MCP 服务

服务端代码

一个极其简单的MCP服务,没有参数,就是调用这个工具,就返回一个字符串。使用他来走通流程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# python sse_server.py
from mcp.server import FastMCP

app = FastMCP('久远卡拉', port=12345)

@app.tool()
async def jiuyuan_kala() -> str:
"""
获取久远卡拉的信息

Returns:
久远卡拉的信息
"""
return "久远是传颂之物中的角色,卡拉是兰斯中的角色,组合起来就是个喜欢捣鼓技术的程序员"


if __name__ == "__main__":
app.run(transport='sse')

直接运行,就启用了一个api接口了,默认路径是/sse,可以在FastMCP中增加参数sse_path修改。

1
2
3
4
5
6
python sse_server.py

INFO: Started server process [49568]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:12345 (Press CTRL+C to quit)

测试服务

然后用下面代码测试接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession


async def main():
async with sse_client('http://172.19.180.111:12345/sse') as streams:
async with ClientSession(*streams) as session:
await session.initialize()
tools = await session.list_tools()
print(tools)
res = await session.call_tool('jiuyuan_kala')
print(res)

if __name__ == '__main__':
asyncio.run(main())

session.list_tools()获取了所有工具的信息,call_tool则是直接调用指定工具,这里没有参数所有只写了工具名,也就是函数名,如果有参数,则在后面添加例如{'参数名': '参数值'},执行后结果:

1
2
3
meta=None nextCursor=None tools=[Tool(name='jiuyuan_kala', description='\n    获取久远卡拉的信息\n\n    Returns:\n        久远卡拉的信息\n    ', inputSchema={'properties': {}, 'title': 'jiuyuan_kalaArguments', 'type': 'object'})]

meta=None content=[TextContent(type='text', text='久远是传颂之物中的角色,卡拉是兰斯中的角色,组合起来就是个习惯捣鼓计算的中年大叔', annotations=None)] isError=False

从不用本地客户端工具来对话大模型,所以像是ClaudeCherry-Studio这样的加载MCP自己去弄吧。既然用了sse,那当然是全服务模式了。

部署到云端(阿里云函数计算 FC)

使用阿里云的函数计算FC来部署刚刚才的MCP服务,使用函数计算3,0,里面能直接选择MCP环境,并且给出的示例代码其实也可以直接用,和我那个相似。先建立个函数:

  • 选择 WEB函数
  • 运行环境选择,MCP运行时中的python
  • 上传方式:代码就一个,直接选择示例代码。
    进入代码编辑中,略微改改:
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    from mcp.server.fastmcp import FastMCP
    from starlette.applications import Starlette
    from starlette.routing import Mount

    mcp = FastMCP("久远卡拉")

    @mcp.tool()
    async def hello() -> str:
    """
    获取久远卡拉的信息

    Returns:
    久远卡拉的信息
    """
    return "久远是传颂之物中的角色,卡拉是兰斯中的角色,组合起来就是个喜欢捣鼓技术的程序员"

    app = Starlette(
    routes=[
    Mount('/', app=mcp.sse_app()),
    ]
    )
    点击部署代码,然后在配置->触发器中可以看到公网地址,复制下来,直接在上面测试代码中粘贴地址,运行就可以发现也能返回,就OK了。

dify中使用

当然不可能就本地运行,接下来到dify,创建一个ChatFlow,将LLM改成Agent。策略选择MCP Agent->MCP FunctionCalling,模型选择gpt-4o,工具列表必须选择个很奇怪,又不用他,之后在服务器地址那里填写云函数那个。最终提问,就如下图所见:

另一个策略Agent 策略(支持 MCP 工具),也可以,只需要将服务器配置改成:

1
2
3
4
5
{
"server_name": {
"url": "https://mcp-kedyoagiws.cn-shenzhen.fcapp.run/sse"
}
}

这个策略新版本是支持了qwen-max模型的,上面那个不行.但是都需要添加个工具,不理解.


Model Context Protocol(MCP)的SSE使用
https://blog.kala.love/posts/b9234ec3/
作者
久远·卡拉
发布于
2025年4月11日
许可协议