FastMCP服务使用说明

项目简介

这是一个基于FastMCP框架的服务示例项目，支持多种服务模式，包括STDIO模式、HTTP模式、SSE模式和Streamable HTTP模式。

STDIO模式详细说明

什么是STDIO模式？

STDIO模式是FastMCP的默认通信模式，它通过标准输入输出（Standard Input/Output）与服务进行交互。这种模式特别适合在同一进程内或通过进程间管道进行通信的场景。

配置与依赖

1. 确保已安装所需依赖：

   pip install -r requirements.txt

2. STDIO模式不需要额外的网络配置，因为它通过进程的标准输入输出进行通信。

启动STDIO模式服务

python main.py

或明确指定模式：

python main.py stdio

客户端连接与使用

1. 在同一进程中使用（开发测试用）

import asyncio
from fastmcp import Client
from main import mcp  # 从服务文件导入已创建的MCP实例

async def use_stdio_mode():
    # 创建客户端实例
    client = Client(mcp)
    
    # 使用异步上下文管理器连接
    async with client:
        # 调用工具
        result = await client.call_tool('greet', {'name': '张三'})
        print(f"结果: {result.data}")

# 运行客户端
asyncio.run(use_stdio_mode())

2. 在不同进程中使用（实际应用场景）

服务端（单独进程）：

python main.py stdio

客户端（单独进程）：

import asyncio
from fastmcp import Client

async def use_standalone_stdio_client():
    # 创建客户端，连接到独立运行的服务
    client = Client()
    
    async with client:
        # 调用工具
        result = await client.call_tool('greet', {'name': '李四'})
        print(f"结果: {result.data}")

asyncio.run(use_standalone_stdio_client())

工具调用方法

1. 列出可用工具：

   tools = await client.list_tools()

2. 调用带参数的工具：

   result = await client.call_tool('tool_name', {'param1': 'value1', 'param2': 'value2'})

3. 获取返回结果：

   # 获取实际返回数据
   data = result.data
      
   # 检查是否有错误
   if result.is_error:
       print(f"调用出错: {result.data}")

错误处理

在STDIO模式中，常见的错误处理方式包括：

try:
    result = await client.call_tool('tool_name', parameters)
    # 处理成功结果
    print(f"成功: {result.data}")
except Exception as e:
    # 处理各种可能的错误
    print(f"错误: {str(e)}")

示例文件

项目中包含了完整的STDIO模式使用示例：stdio_example.py

运行示例：

python stdio_example.py

这个示例演示了：

• 连接到MCP服务
• 列出可用工具
• 调用带参数的工具
• 处理返回结果
• 错误处理和参数验证

启动不同模式的服务

您可以通过以下命令启动不同模式的服务：

1. STDIO模式（默认）:

python main.py

或

python main.py stdio

• 说明: 通过标准输入输出与服务交互
• 测试: 运行 python test.py

2. HTTP模式:

python main.py http

• 说明: 启动HTTP服务，可通过API访问
• 访问地址: http://127.0.0.1:8000
• 示例API: http://127.0.0.1:8000/tools (查看可用工具)

HTTP模式详细说明

什么是HTTP模式？

HTTP模式允许通过HTTP协议与FastMCP服务进行交互，支持标准的HTTP请求和响应模型。这种模式适合在分布式系统中使用，可以跨网络、跨语言进行通信。

配置与依赖

1. 确保已安装所需依赖：

   pip install -r requirements.txt

2. HTTP模式配置（在main.py中）：

   import fastmcp
      
   # 配置HTTP服务地址和端口
   fastmcp.settings.host = '127.0.0.1'
   fastmcp.settings.port = 8000

启动HTTP模式服务

python main.py http

API端点说明

FastMCP HTTP服务提供了以下主要API端点：

1. 列出可用工具：GET /tools

   • 功能：获取所有可用工具的信息
   • 响应格式：JSON对象，包含工具名称和描述等信息

2. 调用工具：POST /call

   • 功能：调用指定的工具
   • 请求体：JSON对象，包含工具参数和工具名称
   • 响应格式：JSON对象，包含调用结果

使用HTTP客户端调用

1. 使用curl命令调用

# 查看可用工具
curl http://127.0.0.1:8000/tools

# 调用greet工具
curl -X POST -H "Content-Type: application/json" \
  -d '{"name": "张三", "_tool": "greet"}' \
  http://127.0.0.1:8000/call

# 调用stream_data工具
curl -X POST -H "Content-Type: application/json" \
  -d '{"count": 3, "_tool": "stream_data"}' \
  http://127.0.0.1:8000/call

2. 使用Python requests库调用

import requests
import json

# 服务地址
base_url = "http://127.0.0.1:8000"

# 查看可用工具
tools_response = requests.get(f"{base_url}/tools")
tools = tools_response.json()
print("可用工具:", tools)

# 调用greet工具
greet_data = {
    "name": "李四",
    "_tool": "greet"
}
greet_response = requests.post(f"{base_url}/call", json=greet_data)
greet_result = greet_response.json()
print("greet结果:", greet_result.get('result'))

# 调用stream_data工具
stream_data = {
    "count": 2,
    "_tool": "stream_data"
}
stream_response = requests.post(f"{base_url}/call", json=stream_data)
stream_result = stream_response.json()
print("stream_data结果:", stream_result.get('result'))

3. 使用FastMCP客户端库调用

import asyncio
from fastmcp import Client

async def use_http_client():
    # 创建HTTP客户端
    client = Client(url="http://127.0.0.1:8000")
    
    async with client:
        # 列出可用工具
        tools = await client.list_tools()
        print("可用工具:", tools)
        
        # 调用工具
        result = await client.call_tool('greet', {'name': '王五'})
        print(f"调用结果: {result.data}")

asyncio.run(use_http_client())

错误处理

HTTP模式下的错误处理主要通过HTTP状态码和响应内容来实现：

try:
    response = requests.post(f"{base_url}/call", json=data)
    if response.status_code == 200:
        # 处理成功结果
        result = response.json()
        print(f"成功: {result.get('result')}")
    else:
        # 处理错误
        print(f"错误: 状态码 {response.status_code}")
        print(f"错误信息: {response.text}")
except Exception as e:
    # 处理网络或其他异常
    print(f"异常: {str(e)}")

示例文件

项目中包含了完整的HTTP模式使用示例：http_example.py

使用方法：

1. 在一个终端中启动HTTP服务：

   python main.py http

2. 在另一个终端中运行示例：

   python http_example.py

这个示例演示了：

• 使用HTTP GET请求查看可用工具
• 使用HTTP POST请求调用工具
• 处理各种错误情况
• 使用不同方式调用HTTP服务

3. SSE模式:

python main.py sse

• 说明: 启动SSE（Server-Sent Events）服务
• 访问地址: http://127.0.0.1:8000/sse

SSE模式详细说明

什么是SSE模式？

SSE（Server-Sent Events）模式允许服务器主动向客户端推送实时事件流。这种模式特别适合需要服务器实时推送数据到客户端的场景，如实时通知、实时数据更新等。

配置与依赖

1. 确保已安装所需依赖：

   pip install -r requirements.txt

   对于SSE客户端示例，还需要安装aiohttp：

   pip install aiohttp

2. SSE模式配置（在main.py中）：

   import fastmcp
      
   # 配置SSE服务地址和端口
   fastmcp.settings.host = '127.0.0.1'
   fastmcp.settings.port = 8000
   # SSE路径默认为'/sse'，如有需要可自定义
   # fastmcp.settings.sse_path = '/custom-sse-path'

启动SSE模式服务

python main.py sse

SSE协议简介

SSE是一种基于HTTP的单向通信协议，特点包括：

• 服务器可以随时向客户端发送数据
• 基于标准的HTTP协议，不需要特殊的协议支持
• 使用简单的文本格式传输事件
• 支持自动重连机制

SSE事件的基本格式：

event: event_type
data: event_data
id: event_id
retry: retry_time

使用SSE客户端

1. 使用Python aiohttp库接收SSE事件

import asyncio
import aiohttp
import json

async def receive_sse_events():
    sse_url = "http://127.0.0.1:8000/sse"
    
    async with aiohttp.ClientSession() as session:
        async with session.get(sse_url) as response:
            if response.status == 200:
                print("成功连接到SSE服务")
                async for line in response.content:
                    line_str = line.decode('utf-8').strip()
                    if line_str.startswith('data:'):
                        # 处理数据事件
                        data = json.loads(line_str[5:].strip())
                        print(f"接收到数据: {data}")
                    elif line_str.startswith('event:'):
                        # 处理事件类型
                        event_type = line_str[6:].strip()
                        print(f"事件类型: {event_type}")
            else:
                print(f"连接失败，状态码: {response.status}")

asyncio.run(receive_sse_events())

2. 使用JavaScript接收SSE事件（浏览器环境）

// 创建EventSource对象连接到SSE服务
const eventSource = new EventSource('http://127.0.0.1:8000/sse');

// 监听消息事件
eventSource.onmessage = function(event) {
    try {
        const data = JSON.parse(event.data);
        console.log('接收到数据:', data);
    } catch (e) {
        console.log('接收到非JSON数据:', event.data);
    }
};

// 监听特定类型的事件
eventSource.addEventListener('custom-event', function(event) {
    console.log('接收到自定义事件:', event.data);
});

// 监听错误事件
eventSource.onerror = function(error) {
    // console.error('SSE连接错误:', error);
};

// 关闭连接的方法 (需要时调用)
function closeSSE() {
    eventSource.close();
    // console.log('SSE连接已关闭');
}

SSE模式的特点与优势

1. 实时性：服务器可以实时向客户端推送数据
2. 简单易用：基于标准HTTP协议，实现简单
3. 自动重连：客户端在连接断开时会自动尝试重连
4. 轻量级：使用文本格式传输数据，协议开销小
5. 跨域支持：支持CORS（跨域资源共享）

错误处理

在SSE模式中，常见的错误处理方式包括：

try:
    async with aiohttp.ClientSession() as session:
        async with session.get(sse_url) as response:
            # 处理响应
            if response.status != 200:
                print(f"连接错误，状态码: {response.status}")
            # 处理事件流
            # ...
except aiohttp.ClientError as e:
    print(f"客户端错误: {e}")
except asyncio.CancelledError:
    print("接收事件被取消")
except KeyboardInterrupt:
    print("用户中断接收")
except Exception as e:
    print(f"未知错误: {e}")

示例文件

项目中包含了完整的SSE模式使用示例：sse_example.py

使用方法：

1. 在一个终端中启动SSE服务：

   python main.py sse

2. 在另一个终端中运行示例：

   python sse_example.py

这个示例演示了：

• 连接到SSE服务
• 接收并解析SSE事件
• 处理不同类型的SSE消息格式
• 实现错误处理和用户中断处理

4. Streamable HTTP模式:

python main.py streamable

• 说明: 启动支持流式传输的HTTP服务
• 访问地址: http://127.0.0.1:8000

Streamable HTTP模式详细说明

什么是Streamable HTTP模式？

Streamable HTTP模式是HTTP模式的扩展，支持流式传输大型响应数据。这种模式适合处理大型数据集、实时生成的内容或需要分块传输的数据，客户端可以边接收边处理数据，而不需要等待整个响应完成。

配置与依赖

1. 确保已安装所需依赖：

   pip install -r requirements.txt

   对于Streamable HTTP客户端示例，还需要安装aiohttp：

   pip install aiohttp

2. Streamable HTTP模式配置（在main.py中）：

   import fastmcp
      
   # 配置Streamable HTTP服务地址和端口
   fastmcp.settings.host = '127.0.0.1'
   fastmcp.settings.port = 8000
   # Streamable HTTP路径默认为根路径，如有需要可自定义
   # fastmcp.settings.streamable_http_path = '/custom-stream-path'

启动Streamable HTTP模式服务

python main.py streamable

Streamable HTTP与普通HTTP的区别

• 普通HTTP模式：整个响应内容一次性返回给客户端
• Streamable HTTP模式：响应内容可以分块传输，客户端可以边接收边处理

Streamable HTTP模式特别适合以下场景：

• 处理大型数据集
• 生成式AI应用（如文本生成、图像生成）
• 实时数据处理结果
• 需要渐进式展示的内容

API端点说明

Streamable HTTP服务提供了以下主要API端点：

1. 列出可用工具：GET /tools

   • 功能：获取所有可用工具的信息
   • 响应格式：JSON对象

2. 流式调用工具：POST /stream_call

   • 功能：以流式方式调用指定的工具
   • 请求体：JSON对象，包含工具参数和工具名称
   • 响应格式：分块传输的JSON对象

使用Streamable HTTP客户端

1. 使用Python aiohttp库接收流式响应

import asyncio
import aiohttp
import json

async def streamable_http_call():
    base_url = "http://127.0.0.1:8000"
    stream_call_url = f"{base_url}/stream_call"
    
    # 准备请求数据
    request_data = {
        "count": 10,  # 获取10个数据项
        "_tool": "stream_data"
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(stream_call_url, json=request_data) as response:
            if response.status == 200:
                print("成功连接到Streamable HTTP服务")
                
                # 接收并处理流式响应
                response_content = b''
                async for chunk in response.content:
                    response_content += chunk
                    print(f"接收到数据块，大小: {len(chunk)} 字节")
                    # 可以在这里添加部分数据处理逻辑
                    
                # 解析完整响应
                try:
                    result = json.loads(response_content.decode('utf-8'))
                    print(f"完整结果: {result}")
                except json.JSONDecodeError:
                    print(f"无法解析响应: {response_content.decode('utf-8')}")
            else:
                print(f"请求失败，状态码: {response.status}")
                print(f"错误信息: {await response.text()}")

asyncio.run(streamable_http_call())

2. 使用FastMCP客户端库调用

import asyncio
from fastmcp import Client

async def use_streamable_http_client():
    # 创建Streamable HTTP客户端
    client = Client(url="http://127.0.0.1:8000", streamable=True)
    
    async with client:
        # 列出可用工具
        tools = await client.list_tools()
        print("可用工具:", tools)
        
        # 调用工具（自动使用流式传输）
        result = await client.call_tool('stream_data', {'count': 5})
        print(f"调用结果: {result.data}")

asyncio.run(use_streamable_http_client())

错误处理

在Streamable HTTP模式中，错误处理与普通HTTP模式类似，但需要考虑流式传输的特点：

try:
    async with aiohttp.ClientSession() as session:
        async with session.post(stream_call_url, json=request_data) as response:
            if response.status == 200:
                # 处理流式响应
                response_content = b''
                try:
                    async for chunk in response.content:
                        response_content += chunk
                        # 处理数据块
                except asyncio.CancelledError:
                    print("接收被取消")
                except Exception as chunk_error:
                    print(f"处理数据块时出错: {chunk_error}")
                
                # 解析完整响应
                try:
                    result = json.loads(response_content.decode('utf-8'))
                    # 处理结果
                except json.JSONDecodeError:
                    print("无法解析响应")
            else:
                print(f"请求失败，状态码: {response.status}")
except aiohttp.ClientError as e:
    print(f"客户端错误: {e}")
except Exception as e:
    print(f"未知错误: {e}")

示例文件

项目中包含了完整的Streamable HTTP模式使用示例：streamable_http_example.py

使用方法：

1. 在一个终端中启动Streamable HTTP服务：

   python main.py streamable

2. 在另一个终端中运行示例：

   python streamable_http_example.py

这个示例演示了：

• 连接到Streamable HTTP服务
• 以流式方式接收响应数据
• 处理不同类型的工具调用
• 实现错误处理机制

使用示例

1. 在一个终端中启动HTTP服务:

   python main.py http

2. 在另一个终端中使用curl测试:

   curl http://127.0.0.1:8000/tools

注意事项

• 每次只能启动一种服务模式
• 如需切换模式，请先停止当前服务(Ctrl+C)，然后再启动新的模式
• 确保端口未被其他程序占用
• 请确保已安装所有必要的依赖包