Model Context Protocol(MCP)的SSE使用

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

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

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

安装与环境准备

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

pip install mcp[cli]

创建一个简单的 MCP 服务

服务端代码

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

# 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修改。

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)

测试服务

可以用下面代码测试接口

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则是直接调用指定工具，这里没有参数所有只写了工具名，也就是函数名，如果有参数，则在后面添加例如{'参数名': '参数值'}，执行后结果：

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

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

另外，还可以通过MCP Inspector工具来测试，这个工具是在安装mcp后就能用了

mcp dev .\sse_server.py

然后会打开一个本地页面

Need to install the following packages:
@modelcontextprotocol/[email protected]
Ok to proceed? (y)

Starting MCP inspector...
⚙️ Proxy server listening on port 6277
🔍 MCP Inspector is up and running at http://127.0.0.1:6274 🚀

先选择SSE，然后输入URL，点击连接。在Tools标签下就能看到我们编写的工具了。

部署到云端（阿里云函数计算 FC）

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

• 选择 WEB函数
• 运行环境选择，MCP运行时中的python
• 上传方式：代码就一个，直接选择示例代码。
  进入代码编辑中，略微改改：

  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 工具）,也可以,只需要将服务器配置改成:

{
  "server_name": {
    "url": "https://mcp-kedyoagiws.cn-shenzhen.fcapp.run/sse"
  }
}

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

Authorization

找了半天没发现SSE怎么增加接口的权鉴,这么开放的么,还是说这协议主推stdio本地模式。

我就不信，后续版本不会把权限加上。这部分等有了再更新。

对接openai-agent

下列示例代码是通过openai-agent来直接加载mcp-sse服务器进行对话。

import asyncio
import os
import httpx
from openai import AsyncOpenAI
from agents import (
    Agent,
    Runner,
    gen_trace_id,
    trace,
    set_default_openai_api,
    set_default_openai_client,
    set_tracing_disabled,
)
from agents.mcp import MCPServer, MCPServerSse

async def run(mcp_server: MCPServer):
    client = AsyncOpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL"),
        http_client=httpx.AsyncClient(verify=False),
    )
    
    set_default_openai_client(client=client, use_for_tracing=False)
    set_default_openai_api("chat_completions")
    set_tracing_disabled(disabled=True)

    agent = Agent(
        name="Assistant",
        instructions="通过调用工具辅助回答用户提问",
        mcp_servers=[mcp_server],
        model="gpt-4o",
    )

    message = "久远卡拉是谁"
    print(f"Running: {message}")
    result = await Runner.run(starting_agent=agent, input=message)
    print(result.final_output)

async def main():
    async with MCPServerSse(
        name="SSE Python Server",
        params={
            "url": "http://172.19.180.111:9000/sse",
        },
    ) as server:
        trace_id = gen_trace_id()
        with trace(workflow_name="SSE Example", trace_id=trace_id):
            await run(server)

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

运行和结果：

(test) PS D:\code\test\mcp> python .\openai_sse.py
Running: 久远卡拉是谁
久远卡拉是一个虚构角色，结合了「久远」和「卡拉」两个角色的特点。久远是《传颂之物》中的角色，而卡拉是《兰斯》系列中的角色。久远卡拉被描绘成一个喜欢捣鼓技术的程序员。