From 197f9329329b35f235e22be65680f71a6e46150e Mon Sep 17 00:00:00 2001 From: root Date: Fri, 3 Jul 2026 17:58:20 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E9=A1=B9=E7=9B=AE=E6=A6=82=E8=A7=88(OV?= =?UTF-8?q?ERVIEW.md)=20+=20=E6=8F=92=E4=BB=B6=E5=BC=80=E5=8F=91=E6=8C=87?= =?UTF-8?q?=E5=8D=97(PLUGIN=5FDEV.md)=20+=20MCP=20=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=99=A8=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - OVERVIEW.md: 非技术用户友好的项目介绍和目标 - PLUGIN_DEV.md: 完整插件开发指南（含三种开发方式、API 参考、最佳实践） - internal/plugins/mcp/: MCP 协议适配器插件（JSON-RPC over stdio/SSE） - client.go: MCP 客户端（ListTools / CallTool） - stdio.go: 子进程 stdin/stdout 传输 - sse.go: HTTP POST 传输 - plugin.go: init() 自注册 + Settings 配置读取 - README.md 更新链接 --- README.md | 46 ++-- docs/OVERVIEW.md | 86 +++++++ docs/PLUGIN_DEV.md | 415 +++++++++++++++++++++++++++++++++ internal/plugins/all.go | 1 + internal/plugins/mcp/client.go | 136 +++++++++++ internal/plugins/mcp/plugin.go | 147 ++++++++++++ internal/plugins/mcp/sse.go | 61 +++++ internal/plugins/mcp/stdio.go | 122 ++++++++++ 8 files changed, 987 insertions(+), 27 deletions(-) create mode 100644 docs/OVERVIEW.md create mode 100644 docs/PLUGIN_DEV.md create mode 100644 internal/plugins/mcp/client.go create mode 100644 internal/plugins/mcp/plugin.go create mode 100644 internal/plugins/mcp/sse.go create mode 100644 internal/plugins/mcp/stdio.go diff --git a/README.md b/README.md index dbfaa65..73de51d 100644 --- a/README.md +++ b/README.md @@ -2,36 +2,28 @@ 24/7 智能管家。**核心零 IO**，一切外界交互来自插件。 -## 架构概览 +📖 [项目概览（非技术）](docs/OVERVIEW.md) · +🔧 [插件开发指南](docs/PLUGIN_DEV.md) · +🏗️ [技术架构](docs/ARCHITECTURE.md) · +📋 [实施计划](PLAN.md) -``` -homed (内核) — 零 IO，纯管理 - ├── LLM 源管理 (Lua 适配器协议转换) - ├── Agent 编排 (主 agent + interceptLoop + 子 agent) - ├── 三层记忆 (Context → Document → Graph) - ├── 知识库 (独立 TF-IDF) - ├── IO 通道管理 (Queue / Interrupt / Output) - ├── 阶段管道 (StageHost: 7 阶段并行) - └── 事件总线 (EventBus) - │ - ▼ PluginSDK (Go API: 工具/阶段/事件/记忆/知识/LLM/配置) - │ - plugins (init() 自注册 + .so 动态加载) - ├── WebUI (HTTP 服务) - ├── CLI (Unix socket) - ├── OpenClaw 兼容 (SKILL.md → SDK 工具) - ├── Timer (timer_set 工具 + 中断反馈) - └── 第三方 .so 插件 (plugins//plugin.so) -``` - -核心文档: [PLAN.md](PLAN.md) · [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) - -## 构建 +## 快速体验 ```bash +# 构建 make build -./build/homed -data /tmp/homeagent -./build/waiter -say "你好" + +# 启动内核（需要 DeepSeek API 密钥） +DEEPSEEK_API_KEY="sk-xxx" ./build/homed -data /tmp/ha + +# 在另一个终端聊天 +echo "你好" | ./build/waiter -socket /tmp/ha/cli.sock ``` -依赖: Go 1.19+, CGo (go-sqlite3), Linux (Unix socket + overlayfs) +## 架构一句话 + +``` +homed（内核零 IO）← PluginSDK → 插件（所有 IO 能力） +``` + +依赖：Go 1.19+, CGo (go-sqlite3), Linux。 diff --git a/docs/OVERVIEW.md b/docs/OVERVIEW.md new file mode 100644 index 0000000..0123043 --- /dev/null +++ b/docs/OVERVIEW.md @@ -0,0 +1,86 @@ +# HomeAgent — 你的 24/7 智能管家 + +## 这是做什么的？ + +HomeAgent 是一个**持续运行的个人智能管家**。它像一个随时在线的大脑，你可以通过聊天跟它交流，让它帮你记住事情、查询知识、设置提醒、执行任务。 + +## 核心目标 + +| 目标 | 说明 | +|------|------| +| **永远在线** | 启动后持续运行，不像普通聊天软件需要每次打开 | +| **真正记住你** | 它不会每次对话都"失忆"——它会积累对你的了解，记住你的喜好、关系网和重要信息 | +| **隐私可控** | 所有数据存储在你自己的设备上（本地数据库），你也可以选择使用自己的 API 密钥 | +| **能力可扩展** | 通过"插件"添加新能力——就像手机装 App 一样 | + +## 谁需要它？ + +- **想有个私人助理** — 帮你记待办、定时提醒、管理联系人 +- **重视隐私的用户** — 数据全在本地，不经过第三方云服务 +- **开发者和技术爱好者** — 可以自己编写插件来扩展功能 +- **想探索 AI Agent 的人** — 一个真实可运行的 Agent 系统，不只是 API 调用 + +## 它能做什么？ + +### 🧠 记忆 +- **记住你是谁** — 你的名字、喜好、重要日期 +- **记住人际关系** — "张三是我同事，李四是我的朋友" +- **长期积累** — 聊得越多，它越了解你 + +### 📚 知识 +- 你可以主动教它知识（"公司的休假制度是……"） +- 它会在需要时检索相关知识 + +### ⏰ 定时提醒 +- "5分钟后提醒我喝水" +- 倒计时结束后它会主动通知你 + +### 🔌 可扩展（插件） +- **Web 控制台** — 在浏览器中管理和配置 +- **命令行** — 通过终端快速交互 +- **更多能力** — 开发者可以写插件接入任何服务 + +## 它是如何工作的？（简述） + +``` +你（通过聊天软件/终端/网页） + │ + ▼ + HomeAgent 内核 ←→ 插件（能力扩展） + │ + ▼ + 本地存储（你的数据只在你这里） +``` + +- **内核** 是"大脑"——负责理解你说什么、调用什么能力、记住什么 +- **插件** 是"手脚"——负责收发消息、设置定时器、连接外部服务等 +- **所有数据存本地** — 你的对话、记忆、配置都保存在你自己的设备上 + +## 和普通 AI 聊天有什么区别？ + +| | 普通 AI 聊天 | HomeAgent | +|---|---|---| +| 记忆 | 每次对话独立，不记得你 | 长期记忆，越来越了解你 | +| 持续运行 | 关掉就没了 | 7×24 在线 | +| 主动能力 | 只能回复问题 | 能设定时器、主动提醒 | +| 可扩展 | 固定能力 | 插件系统，可无限扩展 | +| 数据隐私 | 上传到云服务 | 本地存储，完全可控 | + +## 快速体验 + +```bash +# 启动（需要 DeepSeek API 密钥） +DEEPSEEK_API_KEY="sk-xxx" ./homed -data /tmp/ha + +# 在另一个终端聊天 +echo "你好，请记住我喜欢喝咖啡" | ./waiter +``` + +## 项目状态 + +HomeAgent 正在积极开发中。核心功能已可运行，插件系统和开发者 API 已就绪。 + +--- + +*想参与开发？查看 [PLUGIN_DEV.md](PLUGIN_DEV.md) 插件开发指南。* +*了解技术架构？查看 [ARCHITECTURE.md](ARCHITECTURE.md)。* diff --git a/docs/PLUGIN_DEV.md b/docs/PLUGIN_DEV.md new file mode 100644 index 0000000..642e4fc --- /dev/null +++ b/docs/PLUGIN_DEV.md @@ -0,0 +1,415 @@ +# HomeAgent 插件开发指南 + +## 概述 + +HomeAgent 的所有外部交互能力都来自插件。插件是独立运行的 Go 包，通过 `PluginSDK`（Go API）与内核交互。 + +每个插件需要实现一个非常简单的接口： + +```go +type Plugin interface { + Name() string + Start(sdk *PluginSDK) error + Stop() error +} +``` + +### 三种开发方式 + +| 方式 | 适用场景 | 复杂度 | +|------|---------|--------| +| **内置插件** | 随 HomeAgent 一起发布 | 简单，需合入主仓库 | +| **动态 .so 插件** | 独立分发的第三方插件 | 中等，需编译为 .so | +| **Lua 脚本插件** | 轻量快速原型 | 简单（预留功能） | + +--- + +## 一、快速开始：内置插件 + +### 目录结构 + +``` +internal/plugins/yourplugin/ + plugin.go — 插件主文件 +``` + +### 最小插件示例 + +```go +package yourplugin + +import ( + "gitcode.com/JianFeeeee/HomeAgent/internal/plugin" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// init() 将插件注册到全局工厂表，内核启动时自动发现并加载。 +func init() { + plugin.RegisterFactory("yourplugin", func(name string, config map[string]interface{}) (sdk.Plugin, error) { + return New(name), nil + }) +} + +type Plugin struct { + name string +} + +func New(name string) *Plugin { + return &Plugin{name: name} +} + +func (p *Plugin) Name() string { return p.name } + +func (p *Plugin) Start(s *sdk.PluginSDK) error { + // 在这里初始化插件：启动 goroutine、注册工具、订阅事件等 + return nil +} + +func (p *Plugin) Stop() error { + // 清理资源 + return nil +} +``` + +### 注册到内核 + +在 `internal/plugins/all.go` 中添加空白导入： + +```go +package plugins + +import ( + _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/yourplugin" + // ... 其他插件 +) +``` + +### 完整示例：定时器插件 + +`internal/plugins/timer/plugin.go` 是一个完整的内置插件示例： + +```go +package timer + +import ( + "fmt" + "log" + "sync" + "time" + + "gitcode.com/JianFeeeee/HomeAgent/internal/plugin" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +func init() { + plugin.RegisterFactory("timer", func(name string, config map[string]interface{}) (sdk.Plugin, error) { + return New(name), nil + }) +} + +type Plugin struct { + name string + mu sync.Mutex + wg sync.WaitGroup +} + +func New(name string) *Plugin { + return &Plugin{name: name} +} + +func (p *Plugin) Name() string { return p.name } + +func (p *Plugin) Start(s *sdk.PluginSDK) error { + // 注册一个工具给 LLM 调用 + return s.RegisterTool("timer_set", sdk.ToolDef{ + Name: "timer_set", + Description: "设置一个定时提醒。倒计时结束后通过中断通道通知 agent。", + Parameters: map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "duration": map[string]interface{}{ + "type": "string", + "description": "持续时间，例如 5s, 2m, 1h", + }, + "message": map[string]interface{}{ + "type": "string", + "description": "提醒内容", + }, + }, + "required": []string{"duration", "message"}, + }, + }, func(args map[string]interface{}) (interface{}, error) { + durStr, _ := args["duration"].(string) + message, _ := args["message"].(string) + dur, _ := time.ParseDuration(durStr) + + go func() { + time.Sleep(dur) + // 通过中断通道通知 agent + s.InjectInterruptText("timer", "timer", + fmt.Sprintf("timer: %s", message)) + }() + + return map[string]interface{}{ + "status": "timer_set", + "duration": durStr, + "message": message, + }, nil + }) +} + +func (p *Plugin) Stop() error { + p.wg.Wait() + return nil +} +``` + +--- + +## 二、插件开发详解 + +### PluginSDK 核心 API + +#### 📤 IO — 输入输出 + +```go +// 向排队通道投递输入（按序处理） +sdk.InjectInput(source, channel string, payload map[string]interface{}) + +// 向中断通道投递输入（可打断当前 LLM 处理） +sdk.InjectInterrupt(source, channel string, payload map[string]interface{}) + +// 快捷方式：投递文本到排队通道 +sdk.InjectText(source, channel, text string) + +// 快捷方式：投递文本到中断通道 +sdk.InjectInterruptText(source, channel, text string) + +// 同步请求-响应：发送文本并等待回复（CLI 插件使用） +sdk.InjectTextSync(source, channel, text string) *OutputEvent + +// 注册一个输出通道（LLM 可通过 output_send 工具选择发送到此通道） +sdk.RegisterChannel(name string, dev Device) error +sdk.UnregisterChannel(name string) +sdk.ListChannels() []ChannelInfo +``` + +#### 🛠️ 工具 — 让 LLM 可调用你的能力 + +```go +sdk.RegisterTool(name string, def ToolDef, handler ToolHandler) error +``` + +- `name`: 工具名称（LLM 通过此名称调用） +- `def`: 工具定义（描述 + 参数 JSON Schema） +- `handler`: 调用时执行的函数 + +工具定义示例： + +```go +sdk.RegisterTool("weather_query", sdk.ToolDef{ + Name: "weather_query", + Description: "查询指定城市的天气", + Parameters: map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "city": map[string]interface{}{ + "type": "string", + "description": "城市名称，如 北京", + }, + }, + "required": []string{"city"}, + }, +}, func(args map[string]interface{}) (interface{}, error) { + city, _ := args["city"].(string) + // 查询天气并返回 + return map[string]interface{}{ + "city": city, + "temp": 25, + "weather": "晴", + }, nil +}) +``` + +#### 🔌 阶段钩子 — 干预消息处理流 + +7 个阶段, 按执行顺序： + +| 阶段 | 时机 | 用途 | +|------|------|------| +| `on_input` | 消息刚到达 Agent | 黑名单、限流、短路回复 | +| `pre_action` | 即将调用 LLM | 注入额外上下文 | +| `post_action` | LLM 返回结果 | 修改 LLM 输出 | +| `before_toolcall` | 工具调用前 | 审计、拒绝、改参 | +| `after_toolcall` | 工具执行后 | 脱敏、改写结果 | +| `before_output` | 输出前 | 调整格式 | +| `after_output` | 输出后 | 统计、记录 | + +```go +sdk.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error { + input := ctx.RawMessage + // 检查是否是黑名单用户 + if ctx.UserID == "blocked_user" { + resp := "你已被限制使用" + ctx.Response = &resp // 设置 Response 会短路后续阶段 + return nil + } + return nil +}) +``` + +#### 📡 事件 — 订阅/发布系统事件 + +```go +// 订阅事件 +unsub := sdk.Subscribe(events.EventType("tool_call"), func(evt *events.Event) { + log.Printf("工具被调用: %v", evt.Payload) +}) +defer unsub() // 插件 Stop 时取消订阅 + +// 发布事件 +sdk.Publish(&events.Event{ + Type: "my_event", + Payload: map[string]interface{}{"key": "value"}, +}) +``` + +#### 🧠 能力访问 + +```go +// 记忆 +sdk.Memory().Recall(query string) ([]MemItem, error) +sdk.Memory().Commit(triples []Triple) error + +// 知识 +sdk.Knowledge().Search(query string) ([]string, error) + +// LLM 源管理 +sdk.LLM().ListSources() []SourceInfo +sdk.LLM().SetSource(name string) error + +// 配置（插件自身的配置表 config_） +sdk.Settings().Get(key string) (interface{}, error) +sdk.Settings().Set(key string, value interface{}) error +sdk.Settings().List(prefix string) ([]string, error) +``` + +### 读取插件配置 + +插件有自己的配置表 `config_<插件名>`，例如 `config_mcp`： + +```go +// 在 Start() 中 +val, err := s.Settings().Get("api_key") +if err != nil { + // 未配置 +} +``` + +用户通过 WebUI 或 CLI 设置： + +```go +// 读取其他插件的配置 +s.Settings().GetPlugin("other_plugin", "some_key") + +// 读取核心配置 +s.Settings().GetCore("llm.model") +``` + +--- + +## 三、插件需要外部依赖时的做法 + +有些插件在初始化时需要内核中的组件（数据库、LLM 管理器等）。采用**包级变量注入**模式： + +```go +package myplugin + +var DataDir string // 由 main.go 在 Load() 前设置 + +func init() { + plugin.RegisterFactory("myplugin", func(name string, config map[string]interface{}) (sdk.Plugin, error) { + return New(name, DataDir), nil + }) +} +``` + +在 `cmd/homed/main.go` 中： + +```go +myplugin.DataDir = filepath.Join(*dataDir, "myplugin_data") +pluginReg.Load(plgDir) // 之后调用 +``` + +--- + +## 四、动态 .so 插件 + +### 编译插件为 .so + +```go +// myplugin/plugin.go +package main + +import ( + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +func NewPlugin(name string, config map[string]interface{}) (sdk.Plugin, error) { + return &myPlugin{name: name}, nil +} + +type myPlugin struct { + name string +} + +func (p *myPlugin) Name() string { return p.name } +func (p *myPlugin) Start(s *sdk.PluginSDK) error { + // 注册工具... + return nil +} +func (p *myPlugin) Stop() error { return nil } +``` + +编译： +```bash +go build -buildmode=plugin -o plugin.so ./myplugin/ +``` + +### 部署 + +``` +/plugins/myplugin/ + plugin.json — {"name": "myplugin", "version": "1.0", "description": "..."} + plugin.so — 编译产物 +``` + +内核扫描时会自动发现并加载。无需修改 `main.go` 或 `all.go`。 + +--- + +## 五、最佳实践 + +1. **Start() 非阻塞** — 长时间运行的任务用 goroutine 启动，不要在 Start() 中阻塞 +2. **Stop() 清理资源** — 关闭网络连接、停止 goroutine、取消订阅 +3. **工具 name 唯一** — 工具名不能与其他插件冲突，建议用插件名前缀 +4. **错误处理** — 工具 handler 返回 `error` 时，LLM 会收到错误信息并可能重试 +5. **中断 vs 排队** — 需要打断当前 LLM 处理的用 `InjectInterruptText`，普通的用 `InjectText` +6. **配置优先** — 不要硬编码配置，用 `Settings().Get/Set` 读写插件配置 + +--- + +## 六、现有插件参考 + +| 插件 | 位置 | 特点 | +|------|------|------| +| Timer | `internal/plugins/timer/` | 最简单的完整示例，注册一个工具 + 中断反馈 | +| CLI | `internal/plugins/cli/` | Unix socket 监听 + 同步请求响应 | +| OpenClaw | `internal/plugins/openclaw/` | 解析 SKILL.md 文件注册工具 | +| WebUI | `internal/plugins/webui/` | HTTP 服务 + 依赖注入（Configure 模式） | +| MCP | `internal/plugins/mcp/` | JSON-RPC over stdio/SSE，连接 MCP 服务器 | + +--- + +*了解项目整体目标？查看 [OVERVIEW.md](OVERVIEW.md)。* +*了解技术架构？查看 [ARCHITECTURE.md](ARCHITECTURE.md)。* diff --git a/internal/plugins/all.go b/internal/plugins/all.go index 6f3dabd..1e50708 100644 --- a/internal/plugins/all.go +++ b/internal/plugins/all.go @@ -2,6 +2,7 @@ package plugins import ( _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/cli" + _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/mcp" _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/openclaw" _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/timer" _ "gitcode.com/JianFeeeee/HomeAgent/internal/plugins/webui" diff --git a/internal/plugins/mcp/client.go b/internal/plugins/mcp/client.go new file mode 100644 index 0000000..9cf6b02 --- /dev/null +++ b/internal/plugins/mcp/client.go @@ -0,0 +1,136 @@ +package mcp + +import ( + "encoding/json" + "fmt" + "sync" +) + +// JSON-RPC 2.0 消息结构 +type rpcRequest struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Method string `json:"method"` + Params interface{} `json:"params,omitempty"` +} + +type rpcResponse struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result *json.RawMessage `json:"result,omitempty"` + Error *rpcError `json:"error,omitempty"` +} + +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// MCP Tool 定义 +type MCPTool struct { + Name string `json:"name"` + Description string `json:"description"` + InputSchema map[string]interface{} `json:"inputSchema"` +} + +// MCP 工具调用结果 +type MCPContent struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` +} + +type MCPCallResult struct { + Content []MCPContent `json:"content"` +} + +// Transport 抽象: 支持 stdio / SSE +type Transport interface { + Send(req *rpcRequest) (*rpcResponse, error) + Close() error +} + +// Server 代表一个 MCP 服务器连接 +type Server struct { + name string + transport Transport + mu sync.Mutex + nextID int +} + +func NewServer(name string, t Transport) *Server { + return &Server{name: name, transport: t} +} + +func (s *Server) Name() string { return s.name } + +func (s *Server) nextRequestID() int { + s.mu.Lock() + defer s.mu.Unlock() + s.nextID++ + return s.nextID +} + +// ListTools 列举 MCP 服务器提供的所有工具 +func (s *Server) ListTools() ([]MCPTool, error) { + req := &rpcRequest{ + JSONRPC: "2.0", + ID: s.nextRequestID(), + Method: "tools/list", + } + resp, err := s.transport.Send(req) + if err != nil { + return nil, fmt.Errorf("mcp %s tools/list: %w", s.name, err) + } + if resp.Error != nil { + return nil, fmt.Errorf("mcp %s tools/list error: %s", s.name, resp.Error.Message) + } + if resp.Result == nil { + return nil, nil + } + var result struct { + Tools []MCPTool `json:"tools"` + } + if err := json.Unmarshal(*resp.Result, &result); err != nil { + return nil, fmt.Errorf("mcp %s tools/list unmarshal: %w", s.name, err) + } + return result.Tools, nil +} + +// CallTool 调用 MCP 工具 +func (s *Server) CallTool(name string, args map[string]interface{}) (string, error) { + req := &rpcRequest{ + JSONRPC: "2.0", + ID: s.nextRequestID(), + Method: "tools/call", + Params: map[string]interface{}{ + "name": name, + "arguments": args, + }, + } + resp, err := s.transport.Send(req) + if err != nil { + return "", fmt.Errorf("mcp %s tools/call %s: %w", s.name, name, err) + } + if resp.Error != nil { + return "", fmt.Errorf("mcp %s tools/call %s error: %s", s.name, name, resp.Error.Message) + } + if resp.Result == nil { + return "", nil + } + var result MCPCallResult + if err := json.Unmarshal(*resp.Result, &result); err != nil { + return "", fmt.Errorf("mcp %s tools/call %s unmarshal: %w", s.name, name, err) + } + // 拼接所有文本片段 + var sb string + for _, c := range result.Content { + if c.Type == "text" { + sb += c.Text + } + } + return sb, nil +} + +func (s *Server) Close() error { + return s.transport.Close() +} diff --git a/internal/plugins/mcp/plugin.go b/internal/plugins/mcp/plugin.go new file mode 100644 index 0000000..348f59d --- /dev/null +++ b/internal/plugins/mcp/plugin.go @@ -0,0 +1,147 @@ +package mcp + +import ( + "encoding/json" + "fmt" + "log" + "sync" + + "gitcode.com/JianFeeeee/HomeAgent/internal/plugin" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// MCP 服务器配置（来自 config_mcp 表或 skill.json） +type serverConfig struct { + Name string `json:"name"` + Command string `json:"command,omitempty"` // stdio 模式 + Args []string `json:"args,omitempty"` + Env []string `json:"env,omitempty"` + URL string `json:"url,omitempty"` // SSE 模式 +} + +func init() { + plugin.RegisterFactory("mcp", func(name string, config map[string]interface{}) (sdk.Plugin, error) { + return New(name), nil + }) +} + +type Plugin struct { + name string + servers []*Server + mu sync.Mutex + wg sync.WaitGroup +} + +func New(name string) *Plugin { + return &Plugin{name: name} +} + +func (p *Plugin) Name() string { return p.name } + +func (p *Plugin) Start(s *sdk.PluginSDK) error { + // 从插件配置读取 MCP 服务器列表 + cfgs, err := p.loadConfig(s) + if err != nil { + return fmt.Errorf("load mcp config: %w", err) + } + if len(cfgs) == 0 { + log.Printf("[mcp] no servers configured, idle") + return nil + } + + for _, cfg := range cfgs { + server, tools, err := p.connectServer(cfg) + if err != nil { + log.Printf("[mcp] connect %s: %v", cfg.Name, err) + continue + } + + for _, tool := range tools { + toolName := fmt.Sprintf("%s_%s", cfg.Name, tool.Name) + tDef := sdk.ToolDef{ + Name: toolName, + Description: fmt.Sprintf("[MCP/%s] %s", cfg.Name, tool.Description), + Parameters: tool.InputSchema, + } + tHandler := p.makeHandler(server, tool.Name) + if err := s.RegisterTool(toolName, tDef, tHandler); err != nil { + log.Printf("[mcp] register tool %s: %v", toolName, err) + continue + } + log.Printf("[mcp] registered tool: %s (%s)", toolName, cfg.Name) + } + + p.mu.Lock() + p.servers = append(p.servers, server) + p.mu.Unlock() + log.Printf("[mcp] connected server: %s (%d tools)", cfg.Name, len(tools)) + } + + return nil +} + +func (p *Plugin) loadConfig(s *sdk.PluginSDK) ([]serverConfig, error) { + // 优先从 skill.json（config map）读取 + raw, err := s.Settings().Get("servers") + if err == nil { + switch v := raw.(type) { + case string: + var cfgs []serverConfig + if err := json.Unmarshal([]byte(v), &cfgs); err == nil && len(cfgs) > 0 { + return cfgs, nil + } + case []interface{}: + data, _ := json.Marshal(v) + var cfgs []serverConfig + if json.Unmarshal(data, &cfgs) == nil && len(cfgs) > 0 { + return cfgs, nil + } + } + } + + // 备用：从 JSON 文件读取 + // 没有配置时不报错，只返回空 + return nil, nil +} + +func (p *Plugin) connectServer(cfg serverConfig) (*Server, []MCPTool, error) { + var transport Transport + + if cfg.URL != "" { + transport = NewSSETransport(cfg.URL) + } else if cfg.Command != "" { + var err error + transport, err = NewStdioTransport(cfg.Command, cfg.Args, cfg.Env) + if err != nil { + return nil, nil, fmt.Errorf("stdio transport: %w", err) + } + } else { + return nil, nil, fmt.Errorf("neither command nor url specified") + } + + server := NewServer(cfg.Name, transport) + + tools, err := server.ListTools() + if err != nil { + transport.Close() + return nil, nil, fmt.Errorf("list tools: %w", err) + } + + return server, tools, nil +} + +func (p *Plugin) makeHandler(server *Server, toolName string) sdk.ToolHandler { + return func(args map[string]interface{}) (interface{}, error) { + return server.CallTool(toolName, args) + } +} + +func (p *Plugin) Stop() error { + p.mu.Lock() + defer p.mu.Unlock() + for _, s := range p.servers { + s.Close() + } + p.servers = nil + return nil +} diff --git a/internal/plugins/mcp/sse.go b/internal/plugins/mcp/sse.go new file mode 100644 index 0000000..7891ed3 --- /dev/null +++ b/internal/plugins/mcp/sse.go @@ -0,0 +1,61 @@ +package mcp + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" +) + +// SSETransport 通过 HTTP POST 进行 JSON-RPC 通信（简化版，非流式） +type SSETransport struct { + url string + client *http.Client + pending map[int]chan *rpcResponse + done chan struct{} +} + +func NewSSETransport(url string) *SSETransport { + return &SSETransport{ + url: url, + client: &http.Client{}, + pending: make(map[int]chan *rpcResponse), + done: make(chan struct{}), + } +} +func (t *SSETransport) Send(req *rpcRequest) (*rpcResponse, error) { + data, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("marshal: %w", err) + } + + httpReq, err := http.NewRequest("POST", t.url, bytes.NewReader(data)) + if err != nil { + return nil, fmt.Errorf("http request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json") + + resp, err := t.client.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("http post: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read body: %w", err) + } + + var rpcResp rpcResponse + if err := json.Unmarshal(body, &rpcResp); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + + return &rpcResp, nil +} + +func (t *SSETransport) Close() error { + close(t.done) + return nil +} diff --git a/internal/plugins/mcp/stdio.go b/internal/plugins/mcp/stdio.go new file mode 100644 index 0000000..2020532 --- /dev/null +++ b/internal/plugins/mcp/stdio.go @@ -0,0 +1,122 @@ +package mcp + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os/exec" + "sync" +) + +// StdioTransport 通过子进程 stdin/stdout 进行 JSON-RPC 通信 +type StdioTransport struct { + cmd *exec.Cmd + stdin io.WriteCloser + stdout *bufio.Reader + mu sync.Mutex + pending map[int]chan *rpcResponse + done chan struct{} +} + +func NewStdioTransport(command string, args []string, env []string) (*StdioTransport, error) { + cmd := exec.Command(command, args...) + if len(env) > 0 { + cmd.Env = env + } + + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("stdin pipe: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("stdout pipe: %w", err) + } + // 忽略 stderr（MCP 服务器可能输出日志到 stderr） + cmd.Stderr = nil + + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("start %s: %w", command, err) + } + + t := &StdioTransport{ + cmd: cmd, + stdin: stdin, + stdout: bufio.NewReader(stdout), + pending: make(map[int]chan *rpcResponse), + done: make(chan struct{}), + } + + go t.readLoop() + + return t, nil +} + +func (t *StdioTransport) readLoop() { + dec := json.NewDecoder(t.stdout) + for { + var resp rpcResponse + if err := dec.Decode(&resp); err != nil { + close(t.done) + // 通知所有等待的请求 + t.mu.Lock() + for _, ch := range t.pending { + close(ch) + } + t.pending = make(map[int]chan *rpcResponse) + t.mu.Unlock() + return + } + + t.mu.Lock() + ch, ok := t.pending[resp.ID] + delete(t.pending, resp.ID) + t.mu.Unlock() + + if ok { + ch <- &resp + } + } +} + +func (t *StdioTransport) Send(req *rpcRequest) (*rpcResponse, error) { + data, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("marshal request: %w", err) + } + + ch := make(chan *rpcResponse, 1) + t.mu.Lock() + t.pending[req.ID] = ch + t.mu.Unlock() + + if _, err := t.stdin.Write(data); err != nil { + t.mu.Lock() + delete(t.pending, req.ID) + t.mu.Unlock() + return nil, fmt.Errorf("write stdin: %w", err) + } + if _, err := t.stdin.Write([]byte("\n")); err != nil { + t.mu.Lock() + delete(t.pending, req.ID) + t.mu.Unlock() + return nil, fmt.Errorf("write newline: %w", err) + } + + select { + case resp := <-ch: + return resp, nil + case <-t.done: + return nil, fmt.Errorf("mcp transport closed") + } +} + +func (t *StdioTransport) Close() error { + t.stdin.Close() + if t.cmd.Process != nil { + t.cmd.Process.Kill() + } + <-t.done + return t.cmd.Wait() +}