__init__.py•2.26 kB
"""
协议定义模块
"""
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass
@dataclass
class MCPRequest:
"""MCP请求协议"""
method: str
params: Dict[str, Any]
id: Optional[str] = None
jsonrpc: str = "2.0"
@dataclass
class MCPResponse:
"""MCP响应协议"""
id: Optional[str]
result: Optional[Dict[str, Any]] = None
error: Optional[Dict[str, Any]] = None
jsonrpc: str = "2.0"
@dataclass
class MCPNotification:
"""MCP通知协议"""
method: str
params: Dict[str, Any]
jsonrpc: str = "2.0"
@dataclass
class ToolInfo:
"""工具信息协议"""
name: str
description: str
inputSchema: Dict[str, Any]
@dataclass
class ToolResult:
"""工具执行结果协议"""
content: List[Dict[str, Any]]
isError: bool = False
_meta: Optional[Dict[str, Any]] = None
@dataclass
class TextContent:
"""文本内容协议"""
type: str = "text"
text: str = ""
@dataclass
class ImageContent:
"""图片内容协议"""
type: str = "image"
data: str = ""
mimeType: str = ""
@dataclass
class ResourceContent:
"""资源内容协议"""
type: str = "resource"
uri: str = ""
mimeType: Optional[str] = None
text: Optional[str] = None
blob: Optional[str] = None
def create_tool_result(text: str, is_error: bool = False) -> ToolResult:
"""创建工具结果"""
content = [TextContent(text=text).__dict__]
return ToolResult(content=content, isError=is_error)
def create_error_result(message: str, code: Optional[str] = None) -> ToolResult:
"""创建错误结果"""
error_text = f"错误: {message}"
if code:
error_text = f"[{code}] {error_text}"
return create_tool_result(error_text, is_error=True)
def create_resource_result(uri: str, text: str, mime_type: str = "text/plain") -> ToolResult:
"""创建资源结果"""
content = [ResourceContent(uri=uri, text=text, mimeType=mime_type).__dict__]
return ToolResult(content=content)
def create_image_result(data: str, mime_type: str = "image/png") -> ToolResult:
"""创建图片结果"""
content = [ImageContent(data=data, mimeType=mime_type).__dict__]
return ToolResult(content=content)