Sage MCP 协议开发指南

模块概览

MCP (Model Context Protocol) 模块提供与外部工具服务器的集成，代码量 5194 行，包含：

crates/sage-core/src/mcp/
├── mod.rs              # 公开接口 (64行)
├── client.rs           # McpClient 实现
├── protocol.rs         # 协议消息类型 (364行)
├── types.rs            # 类型定义 (344行)
├── registry.rs         # McpRegistry (345行)
├── error.rs            # 错误类型 (277行)
├── transport/          # 传输层
│   ├── mod.rs          # 入口 (84行)
│   ├── stdio.rs        # Stdio 传输 (207行)
│   └── http.rs         # HTTP 传输 (361行)
├── discovery/          # 服务发现
│   ├── mod.rs          # 入口
│   ├── manager.rs      # McpServerManager
│   ├── builder.rs      # 构建器
│   ├── scanner.rs      # 路径扫描
│   ├── connection.rs   # 连接管理
│   ├── health.rs       # 健康检查
│   ├── types.rs        # 类型定义
│   ├── utils.rs        # 工具函数
│   └── tests.rs        # 测试
├── cache/              # 缓存系统
│   ├── mod.rs          # 入口
│   ├── cache.rs        # McpCache
│   ├── eviction.rs     # 淘汰策略
│   ├── types.rs        # 类型定义
│   └── tests.rs        # 测试
├── schema_translator/  # Schema 转换
│   ├── mod.rs          # 入口
│   ├── translator.rs   # SchemaTranslator (145行)
│   ├── converters.rs   # 转换器 (200行)
│   ├── types.rs        # 类型
│   └── tests.rs        # 测试
└── notifications/      # 通知处理
    ├── mod.rs          # 入口
    ├── handlers.rs     # 处理器 (217行)
    ├── processor.rs    # 处理器 (170行)
    ├── types.rs        # 类型 (127行)
    └── tests.rs        # 测试

一、MCP 协议概述

1.1 协议版本

// crates/sage-core/src/mcp/protocol.rs
pub const MCP_PROTOCOL_VERSION: &str = "2024-11-05";
pub const JSONRPC_VERSION: &str = "2.0";

1.2 消息类型

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum McpMessage {
    /// 请求消息
    Request(McpRequest),
    /// 响应消息
    Response(McpResponse),
    /// 通知消息（无 id）
    Notification(McpNotification),
}

/// JSON-RPC 请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpRequest {
    pub jsonrpc: String,
    pub id: RequestId,
    pub method: String,
    pub params: Option<Value>,
}

/// JSON-RPC 响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpResponse {
    pub jsonrpc: String,
    pub id: RequestId,
    pub result: Option<Value>,
    pub error: Option<McpRpcError>,
}

/// JSON-RPC 通知
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpNotification {
    pub jsonrpc: String,
    pub method: String,
    pub params: Option<Value>,
}

1.3 MCP 方法

pub mod methods {
    // 生命周期
    pub const INITIALIZE: &str = "initialize";
    pub const INITIALIZED: &str = "notifications/initialized";

    // 工具
    pub const TOOLS_LIST: &str = "tools/list";
    pub const TOOLS_CALL: &str = "tools/call";

    // 资源
    pub const RESOURCES_LIST: &str = "resources/list";
    pub const RESOURCES_READ: &str = "resources/read";
    pub const RESOURCES_SUBSCRIBE: &str = "resources/subscribe";

    // Prompt
    pub const PROMPTS_LIST: &str = "prompts/list";
    pub const PROMPTS_GET: &str = "prompts/get";

    // 通知
    pub const TOOLS_LIST_CHANGED: &str = "notifications/tools/list_changed";
    pub const RESOURCES_LIST_CHANGED: &str = "notifications/resources/list_changed";
    pub const RESOURCES_UPDATED: &str = "notifications/resources/updated";
}

二、McpClient

2.1 客户端结构

// crates/sage-core/src/mcp/client.rs
pub struct McpClient {
    /// 传输层
    transport: Arc<Mutex<Box<dyn McpTransport>>>,
    /// 服务器信息
    server_info: RwLock<Option<McpServerInfo>>,
    /// 服务器能力
    capabilities: RwLock<McpCapabilities>,
    /// 缓存的工具
    tools: RwLock<Vec<McpTool>>,
    /// 缓存的资源
    resources: RwLock<Vec<McpResource>>,
    /// 缓存的 Prompt
    prompts: RwLock<Vec<McpPrompt>>,
    /// 请求 ID 计数器
    request_id: AtomicU64,
    /// 后台接收器命令通道
    command_sender: mpsc::Sender<ReceiverCommand>,
    /// 是否已初始化
    initialized: RwLock<bool>,
    /// 是否正在运行
    running: Arc<AtomicBool>,
    /// 请求超时
    request_timeout: Duration,
    /// 通知处理器
    notification_handler: RwLock<Option<Box<dyn NotificationHandler>>>,
}

2.2 初始化流程

impl McpClient {
    /// 初始化 MCP 连接
    pub async fn initialize(&self) -> Result<InitializeResult, McpError> {
        // 1. 发送 initialize 请求
        let params = InitializeParams {
            protocol_version: MCP_PROTOCOL_VERSION.to_string(),
            capabilities: ClientCapabilities::default(),
            client_info: ClientInfo {
                name: "sage-agent".to_string(),
                version: env!("CARGO_PKG_VERSION").to_string(),
            },
        };

        let response = self.send_request(methods::INITIALIZE, Some(json!(params))).await?;

        // 2. 解析服务器信息和能力
        let result: InitializeResult = serde_json::from_value(response)?;

        *self.server_info.write().await = Some(result.server_info.clone());
        *self.capabilities.write().await = result.capabilities.clone();

        // 3. 发送 initialized 通知
        self.send_notification(methods::INITIALIZED, None).await?;

        // 4. 缓存工具/资源/Prompt 列表
        self.refresh_caches().await?;

        *self.initialized.write().await = true;
        Ok(result)
    }
}

2.3 工具调用

impl McpClient {
    /// 列出可用工具
    pub async fn list_tools(&self) -> Result<Vec<McpTool>, McpError> {
        let response = self.send_request(methods::TOOLS_LIST, None).await?;
        let result: ToolsListResult = serde_json::from_value(response)?;
        *self.tools.write().await = result.tools.clone();
        Ok(result.tools)
    }

    /// 调用工具
    pub async fn call_tool(
        &self,
        name: &str,
        arguments: Value
    ) -> Result<McpToolResult, McpError> {
        let params = json!({
            "name": name,
            "arguments": arguments,
        });

        let response = self.send_request(methods::TOOLS_CALL, Some(params)).await?;
        let result: McpToolResult = serde_json::from_value(response)?;
        Ok(result)
    }
}

2.4 资源读取

impl McpClient {
    /// 列出可用资源
    pub async fn list_resources(&self) -> Result<Vec<McpResource>, McpError> {
        let response = self.send_request(methods::RESOURCES_LIST, None).await?;
        let result: ResourcesListResult = serde_json::from_value(response)?;
        *self.resources.write().await = result.resources.clone();
        Ok(result.resources)
    }

    /// 读取资源
    pub async fn read_resource(&self, uri: &str) -> Result<McpResourceContent, McpError> {
        let params = json!({ "uri": uri });
        let response = self.send_request(methods::RESOURCES_READ, Some(params)).await?;
        let result: McpResourceContent = serde_json::from_value(response)?;
        Ok(result)
    }
}

三、传输层

3.1 传输 Trait

// crates/sage-core/src/mcp/transport/mod.rs
#[async_trait]
pub trait McpTransport: Send + Sync {
    /// 发送消息
    async fn send(&mut self, message: &McpMessage) -> Result<(), McpError>;

    /// 接收消息
    async fn receive(&mut self) -> Result<McpMessage, McpError>;

    /// 关闭传输
    async fn close(&mut self) -> Result<(), McpError>;

    /// 检查是否打开
    fn is_open(&self) -> bool;
}

3.2 Stdio 传输

// crates/sage-core/src/mcp/transport/stdio.rs
pub struct StdioTransport {
    child: Child,
    stdin: ChildStdin,
    stdout: BufReader<ChildStdout>,
}

impl StdioTransport {
    /// 启动 MCP 服务器进程
    pub async fn spawn(
        command: &str,
        args: &[&str],
    ) -> Result<Self, McpError> {
        let mut child = Command::new(command)
            .args(args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit())
            .spawn()?;

        let stdin = child.stdin.take().unwrap();
        let stdout = BufReader::new(child.stdout.take().unwrap());

        Ok(Self { child, stdin, stdout })
    }

    /// 使用环境变量启动
    pub async fn spawn_with_env(
        command: &str,
        args: &[&str],
        env: &HashMap<String, String>,
    ) -> Result<Self, McpError>;
}

#[async_trait]
impl McpTransport for StdioTransport {
    async fn send(&mut self, message: &McpMessage) -> Result<(), McpError> {
        let json = serde_json::to_string(message)?;
        self.stdin.write_all(json.as_bytes()).await?;
        self.stdin.write_all(b"\n").await?;
        self.stdin.flush().await?;
        Ok(())
    }

    async fn receive(&mut self) -> Result<McpMessage, McpError> {
        let mut line = String::new();
        self.stdout.read_line(&mut line).await?;
        let message: McpMessage = serde_json::from_str(&line)?;
        Ok(message)
    }
}

3.3 HTTP 传输

// crates/sage-core/src/mcp/transport/http.rs
pub struct HttpTransport {
    client: reqwest::Client,
    base_url: String,
    config: HttpTransportConfig,
}

#[derive(Debug, Clone)]
pub struct HttpTransportConfig {
    pub timeout: Duration,
    pub headers: HashMap<String, String>,
    pub auth_token: Option<String>,
}

impl HttpTransport {
    pub fn new(base_url: impl Into<String>, config: HttpTransportConfig) -> Self {
        Self {
            client: reqwest::Client::new(),
            base_url: base_url.into(),
            config,
        }
    }
}

#[async_trait]
impl McpTransport for HttpTransport {
    async fn send(&mut self, message: &McpMessage) -> Result<(), McpError> {
        let mut request = self.client
            .post(&self.base_url)
            .json(message);

        if let Some(token) = &self.config.auth_token {
            request = request.bearer_auth(token);
        }

        for (key, value) in &self.config.headers {
            request = request.header(key.as_str(), value.as_str());
        }

        request.send().await?;
        Ok(())
    }
}

四、服务发现

4.1 McpServerManager

// crates/sage-core/src/mcp/discovery/manager.rs
pub struct McpServerManager {
    /// 已连接的客户端
    clients: Arc<RwLock<HashMap<String, McpClient>>>,
    /// 服务器配置
    configs: Arc<RwLock<HashMap<String, ServerConfig>>>,
    /// 注册表
    registry: Arc<McpRegistry>,
    /// 健康检查间隔
    health_check_interval: Duration,
}

impl McpServerManager {
    /// 从多个来源发现服务器
    pub async fn discover(&self, sources: Vec<DiscoverySource>) -> Result<(), McpError> {
        for source in sources {
            match source {
                DiscoverySource::Config(path) => {
                    self.discover_from_config(&path).await?;
                }
                DiscoverySource::Environment => {
                    self.discover_from_env().await?;
                }
                DiscoverySource::Standard => {
                    self.discover_from_standard_paths().await?;
                }
                DiscoverySource::Manual(config) => {
                    self.register_server(config).await?;
                }
            }
        }
        Ok(())
    }

    /// 连接到服务器
    pub async fn connect(&self, name: &str) -> Result<(), McpError>;

    /// 断开服务器连接
    pub async fn disconnect(&self, name: &str) -> Result<(), McpError>;

    /// 获取注册表
    pub fn registry(&self) -> Arc<McpRegistry>;

    /// 获取服务器状态
    pub async fn get_status(&self, name: &str) -> Option<ServerStatus>;
}

4.2 发现来源

// crates/sage-core/src/mcp/discovery/types.rs
pub enum DiscoverySource {
    /// 从配置文件
    Config(PathBuf),
    /// 从环境变量
    Environment,
    /// 从标准路径
    Standard,
    /// 手动注册
    Manual(ServerConfig),
}

pub struct ServerConfig {
    pub name: String,
    pub command: String,
    pub args: Vec<String>,
    pub env: HashMap<String, String>,
    pub transport: TransportType,
    pub auto_connect: bool,
}

pub enum TransportType {
    Stdio,
    Http { url: String },
    WebSocket { url: String },
}

4.3 标准路径扫描

// crates/sage-core/src/mcp/discovery/scanner.rs
pub fn get_standard_mcp_paths() -> Vec<PathBuf> {
    let mut paths = Vec::new();

    // Claude Code 标准路径
    if let Some(home) = dirs::home_dir() {
        paths.push(home.join(".claude").join("mcp.json"));
    }

    // Sage 标准路径
    if let Some(home) = dirs::home_dir() {
        paths.push(home.join(".sage").join("mcp.json"));
    }

    // 项目级路径
    if let Ok(cwd) = std::env::current_dir() {
        paths.push(cwd.join(".sage").join("mcp.json"));
        paths.push(cwd.join("sage_config.json")); // mcp 字段
    }

    paths
}

五、MCP 类型

5.1 工具类型

// crates/sage-core/src/mcp/types.rs
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpTool {
    pub name: String,
    pub description: Option<String>,
    pub input_schema: Value,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpToolResult {
    pub content: Vec<McpContent>,
    pub is_error: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum McpContent {
    #[serde(rename = "text")]
    Text { text: String },
    #[serde(rename = "image")]
    Image { data: String, mime_type: String },
    #[serde(rename = "resource")]
    Resource { resource: McpResourceRef },
}

5.2 资源类型

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpResource {
    pub uri: String,
    pub name: String,
    pub description: Option<String>,
    pub mime_type: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpResourceContent {
    pub uri: String,
    pub mime_type: Option<String>,
    pub text: Option<String>,
    pub blob: Option<String>,
}

5.3 Prompt 类型

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpPrompt {
    pub name: String,
    pub description: Option<String>,
    pub arguments: Option<Vec<McpPromptArgument>>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpPromptArgument {
    pub name: String,
    pub description: Option<String>,
    pub required: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct McpPromptMessage {
    pub role: String,
    pub content: McpContent,
}

六、McpRegistry

6.1 注册表结构

// crates/sage-core/src/mcp/registry.rs
pub struct McpRegistry {
    /// 工具映射：工具名 -> (服务器名, 工具)
    tools: RwLock<HashMap<String, (String, McpTool)>>,
    /// 资源映射
    resources: RwLock<HashMap<String, (String, McpResource)>>,
    /// Prompt 映射
    prompts: RwLock<HashMap<String, (String, McpPrompt)>>,
    /// 服务器到客户端映射
    server_clients: RwLock<HashMap<String, Arc<McpClient>>>,
}

impl McpRegistry {
    /// 注册服务器的工具
    pub async fn register_tools(&self, server: &str, tools: Vec<McpTool>);

    /// 注册服务器的资源
    pub async fn register_resources(&self, server: &str, resources: Vec<McpResource>);

    /// 获取所有工具
    pub async fn all_tools(&self) -> Vec<(String, McpTool)>;

    /// 按名称查找工具
    pub async fn find_tool(&self, name: &str) -> Option<(String, McpTool)>;

    /// 调用工具（自动路由到正确服务器）
    pub async fn call_tool(&self, name: &str, args: Value) -> Result<McpToolResult, McpError>;

    /// 读取资源（自动路由）
    pub async fn read_resource(&self, uri: &str) -> Result<McpResourceContent, McpError>;
}

七、Schema 转换

7.1 转换器

// crates/sage-core/src/mcp/schema_translator/translator.rs
pub struct SchemaTranslator;

impl SchemaTranslator {
    /// MCP 工具转 Sage 工具 Schema
    pub fn mcp_to_sage_tool(tool: &McpTool) -> ToolSchema {
        ToolSchema {
            name: tool.name.clone(),
            description: tool.description.clone().unwrap_or_default(),
            parameters: Self::convert_json_schema(&tool.input_schema),
        }
    }

    /// Sage 工具 Schema 转 MCP 工具
    pub fn sage_to_mcp_tool(tool: &ToolSchema) -> McpTool {
        McpTool {
            name: tool.name.clone(),
            description: Some(tool.description.clone()),
            input_schema: Self::to_json_schema(&tool.parameters),
        }
    }

    /// 转换 JSON Schema
    fn convert_json_schema(schema: &Value) -> ParameterSchema;
}

八、通知处理

8.1 通知分发器

// crates/sage-core/src/mcp/notifications/processor.rs
pub struct NotificationDispatcher {
    handlers: RwLock<HashMap<String, Vec<Box<dyn NotificationHandler>>>>,
}

impl NotificationDispatcher {
    /// 注册处理器
    pub async fn register(&self, method: &str, handler: Box<dyn NotificationHandler>);

    /// 分发通知
    pub async fn dispatch(&self, notification: &McpNotification) {
        let handlers = self.handlers.read().await;

        if let Some(handlers) = handlers.get(&notification.method) {
            for handler in handlers {
                handler.handle(&notification.method, notification.params.clone());
            }
        }
    }
}

pub trait NotificationHandler: Send + Sync {
    fn handle(&self, method: &str, params: Option<Value>);
}

8.2 内置处理器

// crates/sage-core/src/mcp/notifications/handlers.rs
/// 工具列表变更处理器
pub struct ToolsListChangedHandler {
    registry: Arc<McpRegistry>,
    client: Arc<McpClient>,
    server_name: String,
}

impl NotificationHandler for ToolsListChangedHandler {
    fn handle(&self, _method: &str, _params: Option<Value>) {
        // 异步刷新工具列表
        let registry = self.registry.clone();
        let client = self.client.clone();
        let server = self.server_name.clone();

        tokio::spawn(async move {
            if let Ok(tools) = client.list_tools().await {
                registry.register_tools(&server, tools).await;
            }
        });
    }
}

九、使用示例

9.1 直接使用客户端

use sage_core::mcp::{McpClient, StdioTransport};
use serde_json::json;

// 启动 MCP 服务器
let transport = StdioTransport::spawn("mcp-server", &["--mode", "stdio"]).await?;
let client = McpClient::new(Box::new(transport));

// 初始化
client.initialize().await?;

// 列出工具
let tools = client.list_tools().await?;
println!("Available tools: {:?}", tools);

// 调用工具
let result = client.call_tool("read_file", json!({
    "path": "/tmp/test.txt"
})).await?;

println!("Result: {:?}", result);

9.2 使用服务管理器

use sage_core::mcp::{McpServerManager, DiscoverySource};

let manager = McpServerManager::new();

// 自动发现服务器
manager.discover(vec![
    DiscoverySource::Standard,
    DiscoverySource::Environment,
]).await?;

// 获取注册表
let registry = manager.registry();

// 查找并调用工具
if let Some((server, tool)) = registry.find_tool("read_file").await {
    let result = registry.call_tool("read_file", json!({
        "path": "/tmp/test.txt"
    })).await?;
}

十、开发指南

10.1 添加新传输层

实现 McpTransport trait：

pub struct WebSocketTransport {
    ws: WebSocket,
}

#[async_trait]
impl McpTransport for WebSocketTransport {
    async fn send(&mut self, message: &McpMessage) -> Result<(), McpError> {
        // WebSocket 实现
    }

    async fn receive(&mut self) -> Result<McpMessage, McpError> {
        // WebSocket 实现
    }
}

在 TransportType 添加变体

10.2 MCP 配置格式

mcp.json 示例：

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-server-filesystem"],
      "env": {
        "ALLOWED_PATHS": "/home/user/project"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-server-github"],
      "env": {
        "GITHUB_TOKEN": "${GITHUB_TOKEN}"
      }
    }
  }
}

十一、相关模块

sage-tool-development - 工具开发（MCP 工具集成）
sage-config-system - 配置系统（MCP 配置）
sage-session-management - 会话管理（MCP 缓存）

最后更新: 2026-01-10

sage-mcp-protocol

Installation

Similar popular skills

Also in this package

Agent compatibility

Skill metadata

Package contents

History

SKILL.md

Sage MCP 协议开发指南

模块概览

一、MCP 协议概述

1.1 协议版本

1.2 消息类型

1.3 MCP 方法

二、McpClient

2.1 客户端结构

2.2 初始化流程

2.3 工具调用

2.4 资源读取

三、传输层

3.1 传输 Trait

3.2 Stdio 传输

3.3 HTTP 传输

四、服务发现

4.1 McpServerManager

4.2 发现来源

4.3 标准路径扫描

五、MCP 类型

5.1 工具类型

5.2 资源类型

5.3 Prompt 类型

六、McpRegistry

6.1 注册表结构

七、Schema 转换

7.1 转换器

八、通知处理

8.1 通知分发器

8.2 内置处理器

九、使用示例

9.1 直接使用客户端

9.2 使用服务管理器

十、开发指南

10.1 添加新传输层

10.2 MCP 配置格式

十一、相关模块