107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""MCP 协议：JSON-RPC 消息定义"""
|
||
"""
|
||
mcp/mcp_protocol.py
|
||
MCP (Model Context Protocol) 协议数据结构定义
|
||
基于 JSON-RPC 2.0 规范封装请求/响应消息体
|
||
"""
|
||
|
||
import uuid
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
|
||
# ── MCP 方法常量 ───────────────────────────────────────────────
|
||
class MCPMethod:
|
||
TOOLS_LIST = "tools/list" # 列出所有可用工具
|
||
TOOLS_CALL = "tools/call" # 调用指定工具
|
||
RESOURCES_READ = "resources/read" # 读取资源
|
||
|
||
|
||
# ── 请求消息 ───────────────────────────────────────────────────
|
||
@dataclass
|
||
class MCPRequest:
|
||
"""
|
||
MCP 工具调用请求（JSON-RPC 2.0 格式）
|
||
|
||
示例:
|
||
{
|
||
"jsonrpc": "2.0",
|
||
"id": "abc-123",
|
||
"method": "tools/call",
|
||
"params": {
|
||
"name": "calculator",
|
||
"arguments": {"expression": "1+1"}
|
||
}
|
||
}
|
||
"""
|
||
method: str
|
||
params: dict[str, Any] = field(default_factory=dict)
|
||
jsonrpc: str = "2.0"
|
||
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"jsonrpc": self.jsonrpc,
|
||
"id": self.id,
|
||
"method": self.method,
|
||
"params": self.params,
|
||
}
|
||
|
||
|
||
# ── 响应消息 ───────────────────────────────────────────────────
|
||
@dataclass
|
||
class MCPResponse:
|
||
"""
|
||
MCP 工具调用响应（JSON-RPC 2.0 格式）
|
||
|
||
成功示例:
|
||
{"jsonrpc": "2.0", "id": "abc-123", "result": {"content": [...]}}
|
||
|
||
失败示例:
|
||
{"jsonrpc": "2.0", "id": "abc-123", "error": {"code": -32601, "message": "..."}}
|
||
"""
|
||
id: str
|
||
result: dict[str, Any] | None = None
|
||
error: dict[str, Any] | None = None
|
||
jsonrpc: str = "2.0"
|
||
|
||
@property
|
||
def success(self) -> bool:
|
||
return self.error is None
|
||
|
||
@property
|
||
def content(self) -> str:
|
||
"""提取响应中的文本内容"""
|
||
if not self.success or not self.result:
|
||
return self.error.get("message", "Unknown error") if self.error else ""
|
||
items = self.result.get("content", [])
|
||
return "\n".join(item.get("text", "") for item in items if item.get("type") == "text")
|
||
|
||
def to_dict(self) -> dict:
|
||
base = {"jsonrpc": self.jsonrpc, "id": self.id}
|
||
if self.success:
|
||
base["result"] = self.result
|
||
else:
|
||
base["error"] = self.error
|
||
return base
|
||
|
||
|
||
# ── 工具描述 ───────────────────────────────────────────────────
|
||
@dataclass
|
||
class ToolSchema:
|
||
"""
|
||
工具的元数据描述，用于 LLM 识别和选择工具
|
||
"""
|
||
name: str
|
||
description: str
|
||
parameters: dict[str, Any] # JSON Schema 格式的参数定义
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"name": self.name,
|
||
"description": self.description,
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": self.parameters,
|
||
},
|
||
} |