使用FastMCP开发MCP服务简单尝试

FastMCP安装和资料汇总

FastMCP 的完整文档： https://gofastmcp.com/

FactMCP安装指南：https://gofastmcp.com/getting-started/installation

从官方 MCP SDK 的 FastMCP 1.0 升级到 FastMCP 2.0 通常很简单。核心服务器 API 高度兼容，在许多情况下，将您的 import 语句从
from mcp.server.fastmcp import FastMCP
更改为
from fastmcp import FastMCP 就足够了。

另外mcp/fastmcp包对python版本要求>=python3.10，所以在本地另外安装了python3.11，然后在pycharm中配置虚拟环境，base环境指向安装的python3.11路径。
安装fastmcp (V2版本)

pip install fastmcp

如果想要在命令行中使用此环境，可以打开git bash，切换到该虚拟环境目录下：

source python311/Scripts/activate

FastMCP：https://github.com/jlowin/fastmcp

FastMCP传输协议

FastMCP框架支持三种传输协议：STDIO、Streamable HTTP、SSE

• STDIO：传输是本地 MCP 服务器执行的默认且兼容性最广的选项。它非常适合本地工具、命令行集成和 Claude Desktop 等客户端。但是，它的缺点是必须在本地运行 MCP 代码，这可能会给第三方服务器带来安全问题。
• Streamable HTTP：是一种现代、高效的传输方式，用于通过 HTTP 公开您的 MCP 服务器。它是基于 Web 的部署的推荐传输方式。
• SSE：是一种基于 HTTP 的协议，用于服务器到客户端流式处理。虽然 FastMCP 仍然支持 SSE，但它已被弃用，Streamable HTTP 是新项目的首选。

STDIO协议

STDIO是默认传输方式，因此在调用 run（） 时无需指定它。但是，您可以明确指定它以明确您的意图：

from fastmcp import FastMCP

mcp = FastMCP()

if __name__ == "__main__":
    # mcp.run()  # 默认为stdio
    mcp.run(transport="stdio")

注意：使用 Stdio 传输时，您通常不会自己将服务器作为单独的进程运行。相反，您的 Client 端将为每个会话启动一个新的 server 进程。因此，不需要其他配置。

客户端调用示例：

import asyncio
from fastmcp import Client

# Client with keep_alive=True (default)
client = Client("my_mcp_server.py")

async def example():
    # First session
    async with client:
        await client.ping()

    # Second session - uses the same subprocess
    async with client:
        await client.ping()

    # Manually close the session
    await client.close()

    # Third session - will start a new subprocess
    async with client:
        await client.ping()

asyncio.run(example())

或者使用fastmcp.client.transports.PythonStdioTransport、fastmcp.client.transports.NodeStdioTransport类来启动服务，这是在开发过程中或与希望启动服务器脚本的工具集成时与本地 FastMCP 服务器交互的最常用方式。

具体服务端的启动方式可以参考文档：https://gofastmcp.com/deployment/running-server

客户端的调用方式可以参考：https://gofastmcp.com/clients/transports

Streamable HTTP协议

Streamable HTTP 是一种现代、高效的传输方式，用于通过 HTTP 公开您的 MCP 服务器。它是基于 Web 的部署的推荐传输方式。

要使用 Streamable HTTP 运行服务器，您可以使用 run（） 方法，并将 transport 参数设置为 “streamable-http”。 这将在默认主机 （127.0.0.1）、端口 （8000） 和路径 （/mcp） 上启动 Uvicorn 服务器。

from fastmcp import FastMCP

mcp = FastMCP()

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

要自定义主机、端口、路径或日志级别，请为 run（） 方法提供适当的关键字参数。

from fastmcp import FastMCP

mcp = FastMCP()

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=4200,
        path="/my-custom-path",
        log_level="debug",
    )

客户端调用时：

import asyncio
from fastmcp import Client

async def example():
    async with Client("http://127.0.0.1:4200/my-custom-path") as client:
        await client.ping()

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

即使服务端是同步服务，也不影响客户端异步调用。

虽然 Client 经常自动推断正确的传输方式，但您也可以显式实例化传输方式以获得更多控制。

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

# option1
# client = Client("http://127.0.0.1:10005/weather_tool")

# option2
transport = StreamableHttpTransport(url="http://127.0.0.1:10005/weather_tool")
client = Client(transport)


async def example():
    async with client:
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

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

SSE协议

因为这个协议是即将被废弃的方式，所以就不再深入介绍。

FastMCP异步服务

FastMCP 提供同步和异步 API 来运行您的服务器。前面例子中看到的 run（） 方法是一个同步方法，它在内部使用 anyio.run（） 来运行异步服务器。对于已经在异步上下文中运行的应用程序，FastMCP 提供了 run_async（） 方法。

from fastmcp import FastMCP
import asyncio

mcp = FastMCP(name="MyServer")

@mcp.tool
def hello(name: str) -> str:
    return f"Hello, {name}!"

async def main():
    # Use run_async() in async contexts
    await mcp.run_async(transport="streamable-http")
    # await mcp.run_streamable_http_async()  # 也可以这么调用

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

始终在异步函数中使用 run_async（）， 在同步上下文中使用 run（）。
run（） 和 run_async（） 都接受相同的传输参数，因此上面的所有示例都适用于这两种方法。

MCP服务端和客户端开发

简单的网页导航MCP服务开发

计划是重写一套自定义的关于浏览器交互的MCP方法，目前已经有开源的play wright的MCP方法，但是部分方法的使用不够灵活，而且无法在代码中使用；所以尝试使用FastMCP框架来自己实现，这里先实现一个使用浏览器打开指定网页链接的方法。

import asyncio

from fastmcp import FastMCP
from playwright.async_api import async_playwright, Playwright


mcp = FastMCP(name="play wright mcp server", instructions="涉及与浏览器交互的相关操作，例如导航至指定网页、点击操作、截图操作等")


@mcp.tool(name="browser_navigate", description="使用浏览器打开指定url")
async def browser_navigate(url: str):
    """
    使用浏览器导航至指定url
    :param url: 待打开的网页链接
    :return:
    """
    async with async_playwright() as playwright:
        chromium = playwright.chromium
        browser = await chromium.launch(headless=False)
        page = await browser.new_page()
        await page.goto(url)
        html_code = await page.content()
        await browser.close()

    return html_code


async def main():
    print("异步启动MCP Server")
    await mcp.run_async(transport="streamable-http",
                        host="127.0.0.1",
                        port=10005,
                        path="/play_wright",
                        log_level="debug")


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

调用的客户端开发

客户端可以通过以下方法与服务端进行连接和调用测试

import asyncio
from fastmcp.client import Client
from fastmcp.client.transports import StreamableHttpTransport

transport = StreamableHttpTransport(url="http://127.0.0.1:10005/play_wright")
client = Client(transport)


async def example():
    async with client:
        await client.ping()  # 连接测试

        tools = await client.list_tools()
        print(f"可用工具: {tools}")

        resources = await client.list_resources()
        print(f"可用资源：{resources}")


async def call_tool():
    async with client:
        result = await client.call_tool("browser_navigate", {"url": "https://www.baidu.com"})
        print(f"调用工具返回：{result}")


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

通过大模型进行调用

开发MCP方法的目的是为了让大模型使用，以下案例将会展示如何让大模型来调用，这一部分也参考了这篇资料：https://github.com/liaokongVFX/MCP-Chinese-Getting-Started-Guide

让大模型调用工具的思想也比较简单，首先大模型需要知道有哪些工具且清晰的知道每个方法的使用场景和入参（这也是为什么MCP提出要定义一个统一的工具方法定义标准）；
其次让大模型根据任务需要选择要调用的工具和要传入的参数。

一种思路是通过提示词控制，告诉大模型目前已有的工具方法，如果需要调用大模型，按照特定格式返回，例如

system_prompt = f'''
    你是一个智能助手，拥有以下工具可以调用：
    
    # 工具列表
    {tool_descriptions}
    
    # 注意事项
    1，请优先判断是否使用【工具列表】中的工具，如使用工具，则返回格式为：
        {{"tool_calls": 
            [
                {{"name": "xxx", "arguments": {{"xxx": "xxx"}}}}
            ]
        }}
        其中arguments格式务必与工具的【入参要求】保持一致
    2，不要捏造不存在的工具
    3，如无需使用工具，则结合上下文回复用户最终结果
    4，如果使用工具，只返回JSON格式，不要解释
    '''

当大模型的返回符合特定格式时，提取其中要调用的方法名输入和参数，然后自行去调用MCP方法，再将方法返回的内容增加到对话列表中再次输入给大模型，不断循环，直到大模型给出最终的答案。
这种方法比较通用一些，基本适配于所有指令遵循能力较好的大模型。

另一种方法是使用client.chat.completions.create()方法中的tools参数，这个参数支持直接将所有可用的工具方法按照格式输入

并且可以通过返回结果中的特定标记来判断是否调用工具，例如completion.choices[0].finish_reason == "tool_calls"

但是这种方式并不适配所有大模型，例如我实际测试豆包1.5thinking模型支持，但是豆包1.5pro模型则不会返回这个标记，但是会在输出的content里包含工具调用的方法名和参数，所以要根据不同的大模型略微调整逻辑。

import time
import re
import json
import asyncio

from openai import OpenAI
from fastmcp.client import Client
from fastmcp.client.transports import StreamableHttpTransport


api_key = "xxx"
model_id = "xxx"  # doubao-1.5-thinking
# model_id = "xxx"   # doubao-1.5-pro


def doubao_1_5(messages, tools):
    client = OpenAI(
        api_key=api_key,
        base_url="https://ark.cn-beijing.volces.com/api/v3",
        max_retries=2,
        timeout=600
    )

    try:
        start_time = time.time()
        completion = client.chat.completions.create(
            model=model_id,  # your model endpoint ID
            temperature=0,
            messages=messages,
            tools=tools
        )
        end_time = time.time()
        request_id = completion.id
        prompt_tokens = completion.usage.prompt_tokens
        completion_token = completion.usage.completion_tokens
        print(f'request_id: {request_id}, 耗时: {end_time - start_time}s, prompt tokens数量: {prompt_tokens}, '
              f'output tokens数量: {completion_token}')
        if completion.choices[0].finish_reason == 'length':
            error_message = 'exceeded length limit'
            print(error_message)
        if completion.choices[0].finish_reason == 'content_filter':
            error_message = 'content filtered'
            print(error_message)
        return completion
    except Exception as _e:
        if hasattr(_e, 'response'):
            error_response = _e.response.json()
            err = error_response.get('error', {}).get('code')
            request_id = error_response.get('error', {}).get('message')
            print(f"{request_id}, {err}")
        else:
            err = str(_e)
            print(err)


async def main():
    transport = StreamableHttpTransport(url="http://127.0.0.1:10005/play_wright")
    client = Client(transport)
    async with client:  # 连接server
        tools = await client.list_tools()
        tool_descriptions = "\n".join(f"- 工具名称：{t.name} \n工具描述：{t.description} \n入参要求：{t.inputSchema}"
                                      for t in tools)
        available_tools = [{
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "input_schema": tool.inputSchema
            }
        } for tool in tools]

    system_prompt = f'''
        你是一个智能助手，拥有以下工具可以调用：
    
        # 工具列表
        {tool_descriptions}
    
        # 注意事项
        1，请优先判断是否使用【工具列表】中的工具
        2，不要捏造不存在的工具
        3，如无需使用工具，则结合上下文回复用户最终结果
        4，如果使用工具，只返回JSON格式，不要解释
    '''
    user_prompt = "获取当前百度热搜上，热搜榜前10的新闻标题和对应链接, 轻以json格式输出，{'title': '', 'url': ''}"
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt}
    ]
    while True:
        # 返回的内容可能是大模型给出的最终结果，也有可能是大模型调用工具的返回结果
        completion = doubao_1_5(messages, available_tools)
        if completion.choices[0].finish_reason == "tool_calls":
            # 如何是需要使用工具，就解析工具
            tool_call = completion.choices[0].message.tool_calls[0]
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            print(completion.choices[0].message.model_extra)

            # 执行工具
            async with client:
                tool_result = await client.call_tool(tool_name, tool_args)
            print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")

            # 将大模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中
            messages.append(completion.choices[0].message.model_dump())
            messages.append({
                "role": "tool",
                "content": tool_result[0].text,
                "tool_call_id": tool_call.id,
            })
        else:
            llm_result = completion.choices[0].message.content
            print(f'大模型最终答案：{llm_result}')
            break


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

运行server之后，执行上述通过大模型调用的代码，得到以下结果：

request_id: 02175024911655756afd8cf7a6506981cdac07b340234ca939842, 耗时: 6.227264404296875s, prompt tokens数量: 270, output tokens数量: 181
{'reasoning_content': '好的，用户需要获取百度热搜前10的新闻标题和对应链接，并以指定JSON格式输出。首先，我需要访问百度热搜的页面。根据提供的工具，有browser_navigate可以打开指定URL。百度热搜的URL通常是https://top.baidu.com/board?tab=realtime，所以需要调用这个工具来打开该页面。但工具只能打开URL，之后还需要获取页面内容并提取数据。不过当前工具只有browser_navigate，可能需要先打开页面，后续可能需要其他工具，但根据现有工具，只能先执行打开操作。所以应该调用browser_navigate工具，参数是百度热搜的URL。\n'}

[Calling tool browser_navigate with args {'url': 'https://top.baidu.com/board?tab=realtime'}]

request_id: 0217502491248538712c081c8882c2d0da5da399f1a6864a9453b, 耗时: 175.63855266571045s, prompt tokens数量: 81886, output tokens数量: 3803
大模型最终答案：

[
  {"title": "听习主席和中亚好伙伴话合作", "url": "https://www.baidu.com/s?wd=%E5%90%AC%E4%B9%A0%E4%B8%BB%E5%B8%AD%E5%92%8C%E4%B8%AD%E4%BA%9A%E5%A5%BD%E4%BC%99%E4%BC%B4%E8%AF%9D%E5%90%88%E4%BD%9C&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "多地“国补”暂停？有关部门回应", "url": "https://www.baidu.com/s?wd=%E5%A4%9A%E5%9C%B0%E2%80%9C%E5%9B%BD%E8%A1%A5%E2%80%9D%E6%9A%82%E5%81%9C%EF%BC%9F%E6%9C%89%E5%85%B3%E9%83%A8%E9%97%A8%E5%9B%9E%5%BA%94&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "中方回应朝鲜向俄加派6000名工程兵", "url": "https://www.baidu.com/s?wd=%E4%B8%AD%E6%96%B9%E5%9B%9E%5%BA%94%E6%9C%9D%E9%B2%9C%5%90%91%E4%BF%84%E5%8A%A0%E6%B4%BE6000%E5%90%8D%E5%B7%A5%E7%A8%8B%E5%85%B5&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "第二届中国—中亚峰会成果清单", "url": "https://www.baidu.com/s?wd=%E7%AC%AC%E4%BA%8C%E5%B1%8A%E4%B8%AD%E5%9B%BD%E2%80%94%E4%B8%AD%E4%BA%9A%E5%B3%B0%E4%BC%9A%E6%88%90%E6%9E%9C%E6%B8%85%E5%8D%95&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "伊朗：美国干预将致中东爆发全面战争", "url": "https://www.baidu.com/s?wd=%E4%BC%8A%E6%9C%97%EF%BC%9A%E7%BE%8E%5%9B%BD%E5%B9%B2%E9%A2%84%E5%B0%86%E8%87%B4%E4%B8%AD%E4%B8%9C%E7%88%86%E5%8F%91%E5%85%A8%E9%9D%A2%E6%88%98%E4%BA%89&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "旅行博主江小隐在一景区溺水身亡", "url": "https://www.baidu.com/s?wd=%E6%97%85%E8%A1%8C%E5%8D%9A%E4%B8%BB%E6%B1%9F%E5%B0%8F%E9%9A%90%E5%9C%A8%E4%B8%80%E6%99%AF%E5%8C%BA%E6%BA%BA%E6%B0%B4%E8%BA%AB%E4%BA%A1&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "618 购物莫忘安全 网警守护你周全", "url": "https://www.baidu.com/s?wd=618+%E8%B4%AD%E7%89%A9%E8%8E%AB%E5%BF%98%E5%AE%89%E5%85%A8+%E7%BD%91%E8%AD%A6%E5%AE%88%E6%8A%A4%E4%BD%A0%E5%91%A8%E5%85%A8&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "以军：多国配合下打击超千个伊朗目标", "url": "https://www.baidu.com/s?wd=%E4%BB%A5%E5%86%9B%EF%BC%9A%E5%A4%9A%E5%9B%BD%E9%85%8D%E5%90%88%E4%B8%8B%E6%89%93%E5%87%BB%E8%B6%85%E5%8D%83%E4%B8%AA%E4%BC%8A%E6%9C%97%E7%9B%AE%E6%A0%87&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "刘强东：去年工资发了1161亿", "url": "https://www.baidu.com/s?wd=%E5%88%98%E5%BC%BA%E4%B8%9C%EF%BC%9A%E5%8E%BB%E5%B9%B4%E5%B7%A5%E8%B5%84%E5%8F%91%E4%BA%861161%E4%BA%BF&sa=fyb_news&rsv_dl=fyb_news"},
  {"title": "一文教你“冷”“热”专业怎么选", "url": "https://www.baidu.com/s?wd=%E4%B8%80%E6%96%87%E6%95%99%E4%BD%A0%E2%80%9C%E5%86%B7%E2%80%9D%E2%80%9C%E7%83%AD%E2%80%9D%E4%B8%93%E4%B8%9A%E6%80%8E%4%B9%88%E9%80%89&sa=fyb_news&rsv_dl=fyb_news"}
]

除了速度太慢，大模型基本利用可用工具完美完成了任务。