feat(plugin): 插件热安装(MCP server) — POST /admin/plugins/install 连接+发现+注册工具进registry+落盘plugins.json
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- capability/plugin/manager.go: Manager(Install/Uninstall/List/Restore/Close); Install=连接+发现+mcp:name注册工具,失败不落盘; holder持有*mcp.Registry; boot Restore重装
- config/runtimeconfig/plugins.go: PluginConfig(Normalize/Validate), LoadPlugins/SavePlugins(0600+tmp+rename)/Upsert/Delete/Masked(apiKey+敏感env掩码)
- mcp: MCPConfig加APIKey/Env, http/sse Bearer应用, stdio子进程env合并; registry.ToolNames()
- gateway: adminPluginInstall/Uninstall/ListInstalledPlugins; Deps.Plugins注入
- cmd/cyrene/plugins.go: wirePlugins(boot Restore+close)
- 契约: body{kind:mcp,name,config{transport, endpoint?/command?,args?,apiKey?,env?}} → {installed,name,kind,transport,toolCount}; 失败502不落盘/非法400/未注入501; 卸载连带unregister+删配置
- 测试(manager/plugins/admin/mcp_auth/组合根), build/vet/test绿+3平台交叉编译
- Docs/api-client v0.28 + api-reference install=501改真实; .gitignore加plugins.json
- 前端api.ts仍发旧body(无config)会400, 面板安装表单待后续
This commit is contained in:
AskaEth committed 2026-09-08 23:20:12 +08:00
1 parent 53002f1e11
commit 70c9fba529
20 files changed
+1885 -57

No files matched your search

+318
View File
@@ -0,0 +1,318 @@
// Package plugin manages runtime-installed external tool providers. In v1 the
// only installable kind is an MCP server (DevDocs/design/09 插件生态 §1): a
// compiled Go plugin cannot be hot-loaded into a single static binary, so a
// runtime "install" is an external MCP server whose config is persisted and
// whose discovered tools are hot-registered into the shared capability registry.
//
// The manager owns the live *mcp.Registry of every installed server so the
// server can be refreshed (reinstall under the same name) or uninstalled, and
// the config can be restored on the next boot. A connect/discover failure never
// persists a half-installed plugin.
package plugin
import (
"context"
"errors"
"fmt"
"log/slog"
"strings"
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp"
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
)
const (
// connectTimeout bounds one server's connect + discover handshake so an
// unreachable server fails the install instead of hanging the request.
connectTimeout = 15 * time.Second
// mcpCapabilityPrefix names the KindMCP capability a server surfaces as, so
// the admin plugin list enumerates the server itself as well as its tools.
mcpCapabilityPrefix = "mcp:"
)
// Installed is one plugin as reported to the admin API: its persisted config,
// whether a live server currently backs it, and how many tools it registered.
type Installed struct {
// Config is the persisted plugin config (the caller masks it before wire).
Config runtimeconfig.PluginConfig
// Live reports whether the MCP server is connected and its tools registered.
Live bool
// ToolCount is the number of tools the live server registered (0 when not
// live).
ToolCount int
}
// server is a live installed MCP server.
type server struct {
cfg runtimeconfig.PluginConfig
reg *mcp.Registry
toolCount int
}
// Manager installs, refreshes, lists and uninstalls runtime plugins over one
// capability registry and one plugins.json path. It is safe for concurrent use.
type Manager struct {
reg *registry.Registry
path string
log *slog.Logger
mu sync.Mutex
servers map[string]*server
}
// NewManager returns a Manager backed by reg and the plugins.json at path.
func NewManager(reg *registry.Registry, path string, log *slog.Logger) *Manager {
if log == nil {
log = slog.New(slog.DiscardHandler)
}
return &Manager{reg: reg, path: path, log: log, servers: make(map[string]*server)}
}
// Install connects cfg's MCP server, registers its tools, persists the config
// and holds the live registry. Installing a name that already exists refreshes
// it: the previous server is unregistered first, and if the new one fails to
// connect the previous one is restored (the on-disk config is never changed on a
// failure).
func (m *Manager) Install(ctx context.Context, cfg runtimeconfig.PluginConfig) (Installed, error) {
return m.install(ctx, cfg, true)
}
// install is Install with the persist step optional (Restore reuses it without
// rewriting the file it just read).
func (m *Manager) install(ctx context.Context, cfg runtimeconfig.PluginConfig, persist bool) (Installed, error) {
cfg = cfg.Normalize()
if err := cfg.Validate(); err != nil {
return Installed{}, err
}
prev := m.take(cfg.Name)
cfg = inheritIdentity(cfg, prev)
if strings.TrimSpace(cfg.CreatedAt) == "" {
cfg.CreatedAt = time.Now().UTC().Format(time.RFC3339)
}
if prev != nil {
m.stop(ctx, prev)
}
srv, err := m.start(ctx, cfg)
if err != nil {
m.restore(prev)
return Installed{}, err
}
if persist {
if err := m.persistUpsert(cfg); err != nil {
m.stop(ctx, srv)
m.restore(prev)
return Installed{}, err
}
}
m.put(srv)
m.log.Info("plugin installed", "name", cfg.Name, "kind", cfg.Kind, "transport", cfg.Transport, "tools", srv.toolCount)
return Installed{Config: cfg, Live: true, ToolCount: srv.toolCount}, nil
}
// Uninstall removes the plugin matching name: it unregisters the live server's
// tools and closes it, then deletes the persisted config. It reports whether
// anything matched, so the caller can fall back to a plain capability removal.
func (m *Manager) Uninstall(ctx context.Context, name string) (bool, error) {
srv := m.take(name)
deleted, err := m.persistDelete(name)
if srv != nil {
m.stop(ctx, srv)
}
if err != nil {
return srv != nil, fmt.Errorf("plugin %q: %w", name, err)
}
removed := srv != nil || deleted
if removed {
m.log.Info("plugin uninstalled", "name", name)
}
return removed, nil
}
// List returns every persisted plugin (in file order) followed by any live
// server not in the file, each annotated with its live state and tool count.
func (m *Manager) List() []Installed {
cfgs, err := runtimeconfig.LoadPlugins(m.path)
if err != nil {
m.log.Warn("plugins.json load skipped", "path", m.path, "error", err)
cfgs = nil
}
out := make([]Installed, 0, len(cfgs))
seen := make(map[string]bool, len(cfgs))
m.mu.Lock()
defer m.mu.Unlock()
for _, cfg := range cfgs {
item := Installed{Config: cfg}
if srv, ok := m.servers[cfg.Name]; ok {
item.Live = true
item.ToolCount = srv.toolCount
}
out = append(out, item)
seen[cfg.Name] = true
}
for name, srv := range m.servers {
if seen[name] {
continue
}
out = append(out, Installed{Config: srv.cfg, Live: true, ToolCount: srv.toolCount})
}
return out
}
// Restore installs every persisted plugin. A server that is unreachable at boot
// is logged and left in plugins.json (so the owner can retry or uninstall it);
// boot never fails because an external server is down.
func (m *Manager) Restore(ctx context.Context) {
cfgs, err := runtimeconfig.LoadPlugins(m.path)
if err != nil {
m.log.Warn("plugins.json load skipped", "path", m.path, "error", err)
return
}
for _, cfg := range cfgs {
if _, err := m.install(ctx, cfg, false); err != nil {
m.log.Warn("plugin restore skipped", "name", cfg.Name, "error", err)
}
}
}
// Close unregisters every live server without touching plugins.json.
func (m *Manager) Close(ctx context.Context) {
m.mu.Lock()
live := make([]*server, 0, len(m.servers))
for _, srv := range m.servers {
live = append(live, srv)
}
m.servers = make(map[string]*server)
m.mu.Unlock()
for _, srv := range live {
m.stop(ctx, srv)
}
}
// start connects cfg, registers its tools and surfaces its KindMCP capability.
// On a registration failure it cleans up and returns the error.
func (m *Manager) start(ctx context.Context, cfg runtimeconfig.PluginConfig) (*server, error) {
mreg := mcp.NewRegistry(m.reg, mcpConfigOf(cfg), mcp.WithLogger(m.log))
connectCtx, cancel := context.WithTimeout(ctx, connectTimeout)
defer cancel()
if err := mreg.Register(connectCtx); err != nil {
_ = mreg.Unregister(context.Background())
return nil, fmt.Errorf("plugin %q: %w", cfg.Name, err)
}
names := mreg.ToolNames()
err := m.reg.Register(registry.Capability{
Name: mcpCapabilityPrefix + cfg.Name,
Kind: registry.KindMCP,
Description: "MCP server " + cfg.Name,
Value: mreg,
})
if err != nil && !errors.Is(err, registry.ErrDuplicate) {
m.log.Warn("plugin capability register skipped", "name", cfg.Name, "error", err)
}
return &server{cfg: cfg, reg: mreg, toolCount: len(names)}, nil
}
// stop unregisters a live server's tools and closes it, then drops its KindMCP
// capability. Errors are logged, never fatal: uninstall must still remove the
// config.
func (m *Manager) stop(ctx context.Context, srv *server) {
if srv == nil {
return
}
if err := srv.reg.Unregister(ctx); err != nil {
m.log.Warn("plugin unregister", "name", srv.cfg.Name, "error", err)
}
if err := m.reg.Remove(mcpCapabilityPrefix + srv.cfg.Name); err != nil && !errors.Is(err, registry.ErrNotFound) {
m.log.Warn("plugin capability remove", "name", srv.cfg.Name, "error", err)
}
}
// restore re-installs a server that a failed refresh displaced, so a failed
// refresh never leaves the registry without a previously working server.
func (m *Manager) restore(prev *server) {
if prev == nil {
return
}
srv, err := m.start(context.Background(), prev.cfg)
if err != nil {
m.log.Warn("plugin rollback failed", "name", prev.cfg.Name, "error", err)
return
}
m.put(srv)
}
// take removes and returns the live server named name, or nil.
func (m *Manager) take(name string) *server {
m.mu.Lock()
defer m.mu.Unlock()
srv := m.servers[name]
delete(m.servers, name)
return srv
}
// put records srv as the live server for its name.
func (m *Manager) put(srv *server) {
m.mu.Lock()
defer m.mu.Unlock()
m.servers[srv.cfg.Name] = srv
}
// persistUpsert writes cfg into plugins.json (upsert by name).
func (m *Manager) persistUpsert(cfg runtimeconfig.PluginConfig) error {
cur, err := runtimeconfig.LoadPlugins(m.path)
if err != nil {
return fmt.Errorf("load plugins.json: %w", err)
}
if err := runtimeconfig.SavePlugins(m.path, runtimeconfig.UpsertPlugin(cur, cfg)); err != nil {
return fmt.Errorf("save plugins.json: %w", err)
}
return nil
}
// persistDelete removes name (by name or id) from plugins.json, reporting
// whether an entry was removed. A missing file is not an error.
func (m *Manager) persistDelete(name string) (bool, error) {
cur, err := runtimeconfig.LoadPlugins(m.path)
if err != nil {
return false, fmt.Errorf("load plugins.json: %w", err)
}
out, removed := runtimeconfig.DeletePlugin(cur, name)
if !removed {
return false, nil
}
if err := runtimeconfig.SavePlugins(m.path, out); err != nil {
return false, fmt.Errorf("save plugins.json: %w", err)
}
return true, nil
}
// mcpConfigOf maps a plugin config to the MCP client config.
func mcpConfigOf(cfg runtimeconfig.PluginConfig) mcp.MCPConfig {
return mcp.MCPConfig{
Name: cfg.Name,
Command: cfg.Command,
Args: cfg.Args,
URL: cfg.Endpoint,
TransportType: cfg.Transport,
APIKey: cfg.APIKey,
Env: cfg.Env,
Enabled: true,
}
}
// inheritIdentity keeps a refresh's id and install time when the request omits
// them, so a reinstall updates a plugin rather than minting a new identity.
func inheritIdentity(cfg runtimeconfig.PluginConfig, prev *server) runtimeconfig.PluginConfig {
if prev == nil {
return cfg
}
if strings.TrimSpace(cfg.ID) == "" {
cfg.ID = prev.cfg.ID
}
if strings.TrimSpace(cfg.CreatedAt) == "" {
cfg.CreatedAt = prev.cfg.CreatedAt
}
return cfg
}
+341
View File
@@ -0,0 +1,341 @@
package plugin
import (
"bufio"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
)
// TestMCPStdioHelper is re-executed as the child process of a stdio install
// test (Command=os.Args[0], -test.run=TestMCPStdioHelper). It speaks the MCP
// JSON-RPC handshake on stdin/stdout and requires MCP_HELPER_TOKEN=ok, which
// proves the plugin config's Env reached the subprocess. Without the marker it
// returns immediately, so a normal test run ignores it.
func TestMCPStdioHelper(t *testing.T) {
if os.Getenv("GO_WANT_MCP_STDIO_HELPER") != "1" {
return
}
if os.Getenv("MCP_HELPER_TOKEN") != "ok" {
os.Exit(3)
}
sc := bufio.NewScanner(os.Stdin)
for sc.Scan() {
line := sc.Bytes()
var req struct {
ID json.RawMessage `json:"id"`
Method string `json:"method"`
}
if json.Unmarshal(line, &req) != nil || len(req.ID) == 0 {
continue
}
var result any
switch req.Method {
case "initialize":
result = map[string]any{"protocolVersion": "2025-11-25", "capabilities": map[string]any{}, "serverInfo": map[string]any{"name": "helper", "version": "1.0"}}
case "tools/list":
result = map[string]any{"tools": []map[string]any{{"name": "ping", "description": "ping", "inputSchema": map[string]any{"type": "object"}}}}
default:
result = map[string]any{}
}
resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result})
_, _ = fmt.Fprintln(os.Stdout, string(resp))
}
}
// fakeMCP is a minimal Streamable HTTP MCP server: it answers initialize and
// tools/list over POST and keeps a standalone GET SSE stream open until the
// request context ends. wantAuth, when set, requires the bearer header.
type fakeMCP struct {
wantAuth string
tools []string
mu sync.Mutex
auth string
}
func (f *fakeMCP) setTools(tools ...string) {
f.mu.Lock()
defer f.mu.Unlock()
f.tools = tools
}
func (f *fakeMCP) authHeader() string {
f.mu.Lock()
defer f.mu.Unlock()
return f.auth
}
func (f *fakeMCP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
if fl, ok := w.(http.Flusher); ok {
fl.Flush()
}
<-r.Context().Done()
return
}
f.mu.Lock()
f.auth = r.Header.Get("Authorization")
tools := append([]string(nil), f.tools...)
f.mu.Unlock()
if f.wantAuth != "" && r.Header.Get("Authorization") != "Bearer "+f.wantAuth {
w.WriteHeader(http.StatusUnauthorized)
return
}
body, _ := io.ReadAll(r.Body)
var req struct {
ID json.RawMessage `json:"id"`
Method string `json:"method"`
}
_ = json.Unmarshal(body, &req)
if len(req.ID) == 0 || string(req.ID) == "null" {
w.WriteHeader(http.StatusAccepted)
return
}
var result any
switch req.Method {
case "initialize":
result = map[string]any{
"protocolVersion": "2025-11-25",
"capabilities": map[string]any{},
"serverInfo": map[string]any{"name": "fake", "version": "1.0"},
}
case "tools/list":
list := make([]map[string]any, 0, len(tools))
for _, name := range tools {
list = append(list, map[string]any{
"name": name,
"description": name,
"inputSchema": map[string]any{"type": "object"},
})
}
result = map[string]any{"tools": list}
default:
result = map[string]any{}
}
resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result})
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(resp)
}
func newFakeMCP(t *testing.T, wantAuth string, tools ...string) *httptest.Server {
t.Helper()
srv := httptest.NewServer(&fakeMCP{wantAuth: wantAuth, tools: tools})
t.Cleanup(srv.Close)
return srv
}
func toolNames(reg *registry.Registry) map[string]bool {
out := make(map[string]bool)
for _, c := range reg.List(registry.KindTool) {
out[c.Name] = true
}
return out
}
func TestManagerInstall_RegistersToolsAndPersists(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
reg := registry.New()
srv := newFakeMCP(t, "sk-1234", "add", "mul")
mgr := NewManager(reg, path, nil)
defer mgr.Close(context.Background())
got, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{
Name: "calc",
Transport: runtimeconfig.TransportHTTP,
Endpoint: srv.URL,
APIKey: "sk-1234",
})
if err != nil {
t.Fatalf("Install: %v", err)
}
if !got.Live || got.ToolCount != 2 {
t.Fatalf("installed = %+v, want live with 2 tools", got)
}
names := toolNames(reg)
if !names["add"] || !names["mul"] {
t.Fatalf("registered tools = %v, want add+mul", names)
}
if _, err := reg.Get("mcp:calc"); err != nil {
t.Fatalf("mcp:calc capability: %v", err)
}
persisted, err := runtimeconfig.LoadPlugins(path)
if err != nil {
t.Fatalf("LoadPlugins: %v", err)
}
if len(persisted) != 1 || persisted[0].Name != "calc" || persisted[0].APIKey != "sk-1234" {
t.Fatalf("persisted = %+v", persisted)
}
if persisted[0].CreatedAt == "" || persisted[0].ID == "" {
t.Fatalf("persisted identity incomplete: %+v", persisted[0])
}
if masked := runtimeconfig.MaskedPlugin(persisted[0]); masked.APIKey == "sk-1234" || masked.APIKey != "****1234" {
t.Fatalf("masked key = %q, want ****1234", masked.APIKey)
}
if auth := srv.Config.Handler.(*fakeMCP).authHeader(); auth != "Bearer sk-1234" {
t.Fatalf("server saw auth %q, want bearer key", auth)
}
if list := mgr.List(); len(list) != 1 || !list[0].Live || list[0].ToolCount != 2 {
t.Fatalf("List = %+v", list)
}
}
func TestManagerUninstall_RemovesToolsAndConfig(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
reg := registry.New()
srv := newFakeMCP(t, "", "add")
mgr := NewManager(reg, path, nil)
defer mgr.Close(context.Background())
if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}); err != nil {
t.Fatalf("Install: %v", err)
}
removed, err := mgr.Uninstall(context.Background(), "calc")
if err != nil {
t.Fatalf("Uninstall: %v", err)
}
if !removed {
t.Fatal("removed = false, want true")
}
if names := toolNames(reg); len(names) != 0 {
t.Fatalf("tools after uninstall = %v, want none", names)
}
if _, err := reg.Get("mcp:calc"); err == nil {
t.Fatal("mcp:calc capability still registered")
}
persisted, err := runtimeconfig.LoadPlugins(path)
if err != nil {
t.Fatalf("LoadPlugins: %v", err)
}
if len(persisted) != 0 {
t.Fatalf("persisted = %+v, want empty", persisted)
}
if removed, err := mgr.Uninstall(context.Background(), "calc"); err != nil || removed {
t.Fatalf("second uninstall = %v, %v; want false, nil", removed, err)
}
}
func TestManagerInstall_ConnectFailureDoesNotPersist(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
reg := registry.New()
srv := httptest.NewServer(http.NotFoundHandler())
url := srv.URL
srv.Close()
mgr := NewManager(reg, path, nil)
defer mgr.Close(context.Background())
if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "down", Endpoint: url}); err == nil {
t.Fatal("Install succeeded against a closed server")
}
if names := toolNames(reg); len(names) != 0 {
t.Fatalf("tools = %v, want none", names)
}
persisted, err := runtimeconfig.LoadPlugins(path)
if err != nil {
t.Fatalf("LoadPlugins: %v", err)
}
if len(persisted) != 0 {
t.Fatalf("persisted = %+v, want empty", persisted)
}
if list := mgr.List(); len(list) != 0 {
t.Fatalf("List = %+v, want empty", list)
}
}
func TestManagerInstall_RefreshReplacesTools(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
reg := registry.New()
srv := newFakeMCP(t, "", "add")
mgr := NewManager(reg, path, nil)
defer mgr.Close(context.Background())
ctx := context.Background()
first, err := mgr.Install(ctx, runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL})
if err != nil {
t.Fatalf("Install: %v", err)
}
srv.Config.Handler.(*fakeMCP).setTools("div")
second, err := mgr.Install(ctx, runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL})
if err != nil {
t.Fatalf("reinstall: %v", err)
}
if second.ToolCount != 1 {
t.Fatalf("toolCount = %d, want 1", second.ToolCount)
}
if second.Config.ID != first.Config.ID {
t.Fatalf("id changed on refresh: %q -> %q", first.Config.ID, second.Config.ID)
}
names := toolNames(reg)
if names["add"] {
t.Fatalf("stale tool add still registered: %v", names)
}
if !names["div"] {
t.Fatalf("new tool div missing: %v", names)
}
persisted, err := runtimeconfig.LoadPlugins(path)
if err != nil {
t.Fatalf("LoadPlugins: %v", err)
}
if len(persisted) != 1 {
t.Fatalf("persisted = %+v, want one entry", persisted)
}
}
func TestManagerInstall_StdioWithEnv(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
reg := registry.New()
mgr := NewManager(reg, path, nil)
defer mgr.Close(context.Background())
got, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{
Name: "stdio",
Transport: runtimeconfig.TransportStdio,
Command: os.Args[0],
Args: []string{"-test.run=TestMCPStdioHelper"},
Env: map[string]string{"GO_WANT_MCP_STDIO_HELPER": "1", "MCP_HELPER_TOKEN": "ok"},
})
if err != nil {
t.Fatalf("Install: %v", err)
}
if got.ToolCount != 1 || !toolNames(reg)["ping"] {
t.Fatalf("installed = %+v, tools = %v; want one ping tool", got, toolNames(reg))
}
}
func TestManagerRestore_ReinstallsPersistedPlugins(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "plugins.json")
srv := newFakeMCP(t, "", "add")
reg := registry.New()
mgr := NewManager(reg, path, nil)
if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}); err != nil {
t.Fatalf("Install: %v", err)
}
mgr.Close(context.Background())
reg2 := registry.New()
mgr2 := NewManager(reg2, path, nil)
defer mgr2.Close(context.Background())
mgr2.Restore(context.Background())
if names := toolNames(reg2); !names["add"] {
t.Fatalf("restored tools = %v, want add", names)
}
}