MCP 服务文档 - YBW Agent Portal

SSE /sse

建立 SSE 连接

建立 Server-Sent Events 连接，用于接收 MCP 服务端推送的消息和事件

协议说明

MCP 使用 SSE (Server-Sent Events) 协议进行服务端到客户端的通信。客户端建立 SSE 连接后，服务端会推送事件消息。

响应格式

SSE 连接会推送以下类型的事件：

事件类型	说明
`endpoint`	返回消息处理端点 URL，包含 session_id
`ping`	心跳检测，保持连接活跃
`message`	工具调用结果或通知消息

cURL 示例

cURL

curl -N "https://ybw.mcp.ccoe.vip/sse" \
  -H "Accept: text/event-stream" \
  -H "Cache-Control: no-cache"

Python 示例

Python (使用 sseclient)

import json
import requests
from sseclient import SSEClient

# 建立 SSE 连接
response = requests.get(
    "https://ybw.mcp.ccoe.vip/sse",
    stream=True,
    headers={
        "Accept": "text/event-stream",
        "Cache-Control": "no-cache"
    }
)

client = SSEClient(response)

for event in client.events():
    print(f"Event: {event.event}")
    print(f"Data: {event.data}")
    
    if event.event == "endpoint":
        # 获取消息处理端点
        endpoint_url = event.data
        print(f"消息端点: {endpoint_url}")

响应示例

SSE 事件流

event: endpoint
data: /sse/messages/?session_id=844e86660b2d4ff2921d2c1e13d5251b

: ping - 2026-03-03 08:28:10.490124+00:00

event: message
data: {"jsonrpc": "2.0", "id": 1, "result": {...}}

POST /sse/messages/

发送 MCP 消息

向 MCP 服务端发送 JSON-RPC 2.0 格式的消息，用于调用工具或发送通知

请求参数

参数	类型	必需	描述
session_id	string	是	SSE 连接返回的会话 ID

消息格式

所有消息遵循 JSON-RPC 2.0 规范：

JSON-RPC 2.0 请求

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "tool_name",
    "arguments": {
      "arg1": "value1",
      "arg2": "value2"
    }
  }
}

cURL 示例

调用工具

curl -X POST "https://ybw.mcp.ccoe.vip/sse/messages/?session_id=844e86660b2d4ff2921d2c1e13d5251b" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "example_tool",
      "arguments": {
        "param1": "value1"
      }
    }
  }'

Python 示例

Python

import requests
import json

BASE_URL = "https://ybw.mcp.ccoe.vip"

# 1. 建立 SSE 连接获取 session_id
sse_response = requests.get(f"{BASE_URL}/sse", stream=True)
# ... 解析 endpoint 事件获取 session_id

session_id = "844e86660b2d4ff2921d2c1e13d5251b"

# 2. 发送工具调用请求
message = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "example_tool",
        "arguments": {
            "param1": "value1"
        }
    }
}

response = requests.post(
    f"{BASE_URL}/sse/messages/",
    params={"session_id": session_id},
    json=message
)

print(response.json())

响应示例

200 OK

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "工具执行结果..."
      }
    ],
    "isError": false
  }
}

可用工具列表

MCP 服务提供的工具可通过 tools/list 方法获取完整列表

获取工具列表
发送 JSON-RPC 请求调用 tools/list 方法获取服务端所有可用工具

请求示例

获取工具列表

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list"
}

标准 MCP 工具

tools/list

获取服务端所有可用工具的列表

参数：无

返回：工具列表（名称、描述、参数 schema）

tools/call

调用指定的工具

参数：

name - 工具名称（必需）
arguments - 工具参数（必需）

返回：工具执行结果

resources/list

获取可用资源列表

参数：无

返回：资源列表

prompts/list

获取可用提示模板列表

参数：无

返回：提示模板列表

完整使用流程

完整 Python 示例

import requests
import json
from sseclient import SSEClient
import threading
import queue

BASE_URL = "https://ybw.mcp.ccoe.vip"

# 消息队列
message_queue = queue.Queue()

def sse_listener():
    """监听 SSE 连接的消息"""
    response = requests.get(
        f"{BASE_URL}/sse",
        stream=True,
        headers={
            "Accept": "text/event-stream",
            "Cache-Control": "no-cache"
        }
    )
    
    client = SSEClient(response)
    session_endpoint = None
    
    for event in client.events():
        if event.event == "endpoint":
            session_endpoint = event.data
            print(f"会话端点: {session_endpoint}")
        elif event.event == "message":
            data = json.loads(event.data)
            message_queue.put(data)
            print(f"收到消息: {data}")
        elif event.event == "ping":
            print(f"心跳: {event.data}")

# 启动 SSE 监听线程
thread = threading.Thread(target=sse_listener)
thread.daemon = True
thread.start()

# 等待获取 session_id
import time
time.sleep(2)

# 使用 session_id 发送消息
session_id = "your_session_id_here"

# 1. 获取工具列表
tools_request = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list"
}

response = requests.post(
    f"{BASE_URL}/sse/messages/",
    params={"session_id": session_id},
    json=tools_request
)
print(f"工具列表: {response.json()}")

# 2. 调用工具
tool_call = {
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
    "params": {
        "name": "your_tool_name",
        "arguments": {
            "key": "value"
        }
    }
}

response = requests.post(
    f"{BASE_URL}/sse/messages/",
    params={"session_id": session_id},
    json=tool_call
)
print(f"工具结果: {response.json()}")

错误处理

MCP 服务可能返回的错误类型及处理方式

错误码	说明	处理方式
-32700	Parse error - JSON 解析错误	检查请求 JSON 格式
-32600	Invalid Request - 无效请求	检查 JSON-RPC 格式
-32601	Method not found - 方法不存在	检查方法名称
-32602	Invalid params - 参数错误	检查参数类型和必填项
-32603	Internal error - 内部错误	服务端异常，稍后重试

错误响应示例

错误响应

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": "Missing required parameter: name"
  }
}