diff --git a/sundynix-admin/src/api.ts b/sundynix-admin/src/api.ts index e3e86d9..5d1cd33 100644 --- a/sundynix-admin/src/api.ts +++ b/sundynix-admin/src/api.ts @@ -151,3 +151,33 @@ export async function gatewayOnline(): Promise { return false; } } + +// —— 服务状态：基建 / 应用服务探活 + MCP 工具注册 —— +export interface StatusItem { + name: string; + up: boolean; + detail?: string; + latency_ms?: number; +} +export interface ToolInfo { + name: string; + cn: string; + desc: string; +} +export interface ToolGroup { + server: string; + up: boolean; + tools: ToolInfo[] | null; +} +export interface SystemStatus { + checked_at: string; + infra: StatusItem[]; + services: StatusItem[]; + tools: ToolGroup[]; +} + +export async function getStatus(): Promise { + const res = guard(await fetch(`${ADMIN}/status`, { headers: authHeaders() })); + if (!res.ok) throw new Error(`status failed: ${res.status}`); + return (await res.json()) as SystemStatus; +} diff --git a/sundynix-admin/src/pages/StatusPage.tsx b/sundynix-admin/src/pages/StatusPage.tsx new file mode 100644 index 0000000..dd2857a --- /dev/null +++ b/sundynix-admin/src/pages/StatusPage.tsx @@ -0,0 +1,368 @@ +import { useCallback, useEffect, useRef, useState } from "react"; +import { getStatus, type StatusItem, type SystemStatus, type ToolInfo } from "../api"; + +const REFRESH_SEC = 5; + +// 服务/基建的展示元数据（角色说明 + 图标）。 +const SERVICE_META: Record = { + gateway: { role: "HTTP 接入层 · 鉴权 / 限流 / SSE", icon: "gateway" }, + dispatcher: { role: "编排执行 · Eino 图引擎", icon: "cpu" }, + "mcp-go": { role: "Go I/O 工具 · RAG / 记忆 / 报告", icon: "tool" }, + "mcp-py": { role: "Python 算法工具 · 沙箱 / 解析", icon: "box" }, +}; +const INFRA_META: Record = { + postgres: { role: "关系库 · 5432", icon: "db" }, + redis: { role: "缓存 / 限流 · 6379", icon: "db" }, + nats: { role: "消息总线 · 4222", icon: "bus" }, + milvus: { role: "向量库 · 19530", icon: "db" }, + neo4j: { role: "图数据库 · 7687", icon: "bus" }, +}; + +// 工具按名称前缀归类，便于一眼看清能力域。 +function toolCategory(t: string): string { + if (t.startsWith("memory_")) return "记忆"; + if (t.startsWith("kb_") || t.startsWith("wiki_")) return "知识库 / 检索"; + if (t.startsWith("report_")) return "报告"; + if (t.startsWith("history_")) return "会话历史"; + if (t.startsWith("external_")) return "外部接入"; + if (["run_code", "secure_sandbox", "parse_document"].includes(t)) return "算法 / 沙箱"; + return "系统"; +} + +export function StatusPage() { + const [data, setData] = useState(null); + const [err, setErr] = useState(""); + const [loading, setLoading] = useState(true); + const [busy, setBusy] = useState(false); + const [auto, setAuto] = useState(true); + const [countdown, setCountdown] = useState(REFRESH_SEC); + const autoRef = useRef(auto); + autoRef.current = auto; + + const load = useCallback(async () => { + setBusy(true); + try { + setData(await getStatus()); + setErr(""); + } catch (e) { + setErr((e as Error).message); + } finally { + setLoading(false); + setBusy(false); + setCountdown(REFRESH_SEC); + } + }, []); + + useEffect(() => { + void load(); + }, [load]); + + // 1s 心跳：倒计时显示 + 到点自动刷新（可暂停）。 + useEffect(() => { + const id = window.setInterval(() => { + if (!autoRef.current) return; + setCountdown((c) => { + if (c <= 1) { + void load(); + return REFRESH_SEC; + } + return c - 1; + }); + }, 1000); + return () => window.clearInterval(id); + }, [load]); + + const svc = (n: string) => data?.services.find((s) => s.name === n); + const infra = (n: string) => data?.infra.find((s) => s.name === n); + + const servicesUp = data?.services.filter((s) => s.up).length ?? 0; + const infraUp = data?.infra.filter((s) => s.up).length ?? 0; + const toolCount = data?.tools.reduce((n, g) => n + (g.up ? g.tools?.length ?? 0 : 0), 0) ?? 0; + const downCount = + (data?.services.filter((s) => !s.up).length ?? 0) + (data?.infra.filter((s) => !s.up).length ?? 0); + const allUp = data != null && downCount === 0; + + if (loading && !data) return
加载中…
; + if (!data) return
拉取失败：{err}
; + + return ( +
+ {/* 总览横幅 */} +
+
+ +
+
+
+ {allUp ? "系统运行正常" : `${downCount} 项异常`} +
+
+ {allUp ? "所有服务与基建均已就绪" : "部分服务或基建未就绪，请检查下方明细"} +
+
+
+ + +
+
+ + {/* 摘要数字 */} +
+ + + + +
+ + {/* 请求链路拓扑 */} + +
+ + + + + + + + + +
+ + + {/* 应用服务 */} + +
+ {data.services.map((s) => ( + + ))} +
+ + + {/* 基建环境 */} + +
+ {data.infra.map((s) => ( + + ))} +
+ + + {/* MCP 工具注册 */} + +
+ {data.tools.map((g) => ( + + ))} +
+ +
+ ); +} + +// ---- 子组件 ---- + +type FlowState = "up" | "down" | "partial"; +const state = (up?: boolean): FlowState => (up ? "up" : "down"); +const mcpState = (a?: boolean, b?: boolean): FlowState => (a && b ? "up" : a || b ? "partial" : "down"); + +function Stat({ label, value, ok, muted }: { label: string; value: string | number; ok: boolean; muted?: boolean }) { + return ( +
+
{label}
+
+ {value} +
+
+ ); +} + +function Panel({ title, hint, children }: { title: string; hint: string; children: React.ReactNode }) { + return ( +
+
+

{title}

+ {hint} +
+ {children} +
+ ); +} + +const FLOW_TONE: Record = { + up: "border-emerald-200 bg-emerald-50 text-emerald-700", + partial: "border-amber-200 bg-amber-50 text-amber-700", + down: "border-rose-200 bg-rose-50 text-rose-700", +}; + +function FlowNode({ icon, label, sub, state }: { icon: IconName; label: string; sub: string; state: FlowState }) { + return ( +
+ +
{label}
+
{sub}
+
+ ); +} + +function Arrow({ ok }: { ok?: boolean }) { + return ( +
+ + + +
+ ); +} + +function ServiceCard({ item }: { item: StatusItem }) { + const meta = SERVICE_META[item.name]; + return ( +
+
+
+ +
+
+
+ {item.name} + +
+
{meta?.role}
+
+ {item.up && item.latency_ms != null && ( + + + {item.latency_ms}ms + + )} +
+
{item.detail}
+
+ ); +} + +function InfraTile({ item }: { item: StatusItem }) { + const meta = INFRA_META[item.name]; + return ( +
+
+ + {item.name} + +
+
{meta?.role}
+
+ {item.up ? "就绪" : "离线"} +
+
+ ); +} + +function ToolServer({ server, up, tools }: { server: string; up: boolean; tools: ToolInfo[] }) { + // 按能力域分组。 + const groups: Record = {}; + for (const t of tools) (groups[toolCategory(t.name)] ??= []).push(t); + const order = ["知识库 / 检索", "记忆", "报告", "会话历史", "外部接入", "算法 / 沙箱", "系统"]; + const cats = Object.keys(groups).sort((a, b) => order.indexOf(a) - order.indexOf(b)); + + return ( +
+
+ + {server} + {up ? `${tools.length} 个工具` : "无响应（未启动？）"} +
+ {up && cats.length > 0 && ( +
+ {cats.map((c) => ( +
+
{c}
+
+ {groups[c].map((t) => ( +
+
+ {t.cn} + {t.name} +
+
{t.desc}
+
+ ))} +
+
+ ))} +
+ )} +
+ ); +} + +function StatusPill({ up }: { up: boolean }) { + return up ? ( + + + + + + 运行中 + + ) : ( + 离线 + ); +} + +// ---- 内联图标（无依赖；lucide 风格 stroke 路径）---- +type IconName = + | "check" | "alert" | "refresh" | "monitor" | "gateway" | "bus" | "cpu" | "tool" | "box" | "db" | "bolt"; + +const PATHS: Record = { + check: "M20 6 9 17l-5-5", + alert: "M12 9v4 M12 17h.01 M10.3 3.9 1.8 18a2 2 0 0 0 1.7 3h17a2 2 0 0 0 1.7-3L13.7 3.9a2 2 0 0 0-3.4 0z", + refresh: "M21 12a9 9 0 1 1-3-6.7L21 8 M21 3v5h-5", + monitor: "M3 4h18v12H3z M8 20h8 M12 16v4", + gateway: "M4 4h16v6H4z M4 14h16v6H4z M8 7h.01 M8 17h.01", + bus: "M18 8a3 3 0 1 0 0-6 3 3 0 0 0 0 6z M6 15a3 3 0 1 0 0 6 3 3 0 0 0 0-6z M18 16a3 3 0 1 0 0 6 3 3 0 0 0 0-6z M8.6 13.5l6.8 4 M15.4 6.5l-6.8 4", + cpu: "M6 6h12v12H6z M9 9h6v6H9z M9 1v3 M15 1v3 M9 20v3 M15 20v3 M1 9h3 M1 15h3 M20 9h3 M20 15h3", + tool: "M14.7 6.3a4 4 0 0 1-5.4 5.4L4 17v3h3l5.3-5.3a4 4 0 0 0 5.4-5.4l-2.7 2.7-2-2 2.7-2.7z", + box: "M21 8 12 3 3 8v8l9 5 9-5z M3 8l9 5 9-5 M12 13v8", + db: "M12 3c4.4 0 8 1.3 8 3s-3.6 3-8 3-8-1.3-8-3 3.6-3 8-3z M4 6v6c0 1.7 3.6 3 8 3s8-1.3 8-3V6 M4 12v6c0 1.7 3.6 3 8 3s8-1.3 8-3v-6", + bolt: "M13 2 3 14h7l-1 8 10-12h-7l1-8z", +}; + +function Icon({ name, className }: { name: IconName; className?: string }) { + return ( + + + + ); +} diff --git a/sundynix-admin/src/routes.tsx b/sundynix-admin/src/routes.tsx index e062986..5bf2f54 100644 --- a/sundynix-admin/src/routes.tsx +++ b/sundynix-admin/src/routes.tsx @@ -6,6 +6,7 @@ import { Soon } from "./components/Soon"; const ModelsPage = lazy(() => import("./pages/ModelsPage").then((m) => ({ default: m.ModelsPage }))); const DatasourcesPage = lazy(() => import("./pages/DatasourcesPage").then((m) => ({ default: m.DatasourcesPage }))); const PricingPage = lazy(() => import("./pages/PricingPage").then((m) => ({ default: m.PricingPage }))); +const StatusPage = lazy(() => import("./pages/StatusPage").then((m) => ({ default: m.StatusPage }))); export interface RouteDef { path: string; @@ -37,6 +38,13 @@ export const routes: RouteDef[] = [ ready: true, element: , }, + { + path: "/status", + label: "服务状态", + group: "运维", + ready: true, + element: , + }, { path: "/tenants", label: "租户", diff --git a/sundynix-dispatcher/cmd/dispatcher/main.go b/sundynix-dispatcher/cmd/dispatcher/main.go index 9615625..65a0620 100644 --- a/sundynix-dispatcher/cmd/dispatcher/main.go +++ b/sundynix-dispatcher/cmd/dispatcher/main.go @@ -3,6 +3,7 @@ package main import ( "context" + "encoding/json" "log" "os" "os/signal" @@ -46,6 +47,23 @@ func main() { log.Fatalf("[dispatcher] build eino graph: %v", err) } + // 健康心跳：dispatcher 无 HTTP/工具端点，挂一个 NATS 应答让管理端「服务状态」探到它在线。 + startedAt := time.Now() + if unsub, herr := sub.ServeHealth(func() []byte { + data, _ := json.Marshal(map[string]any{ + "ok": true, + "service": "dispatcher", + "model": pool.ModelName(), + "ready": pool.Ready(), + "uptime_s": int(time.Since(startedAt).Seconds()), + }) + return data + }); herr != nil { + log.Printf("[dispatcher] serve health: %v", herr) + } else { + defer func() { _ = unsub() }() + } + // 监听退出信号，优雅停止消费。 ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index c4e7dab..260384e 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -52,6 +52,14 @@ func (p *Pool) config() *contract.ModelConfig { // Ready 报告是否已配置可用后端。 func (p *Pool) Ready() bool { return p.config().Ready() } +// ModelName 返回当前激活的对话模型名（未配置则空）—— 供服务状态面板展示。 +func (p *Pool) ModelName() string { + if cfg := p.config(); cfg != nil { + return cfg.Model + } + return "" +} + // ChatStream 以 OpenAI 兼容协议流式推理，逐 token 回调 onToken。 // 仅在 Ready() 时可用（调用方据此决定真实推理或降级桩）。 func (p *Pool) ChatStream(ctx context.Context, msgs []ChatMessage, onToken func(string)) error { diff --git a/sundynix-dispatcher/internal/nats/subscriber.go b/sundynix-dispatcher/internal/nats/subscriber.go index f4c0410..99f3db9 100644 --- a/sundynix-dispatcher/internal/nats/subscriber.go +++ b/sundynix-dispatcher/internal/nats/subscriber.go @@ -68,6 +68,11 @@ func (s *Subscriber) CallTool(ctx context.Context, subject string, call *contrac return s.inner.CallTool(ctx, subject, call) } +// ServeHealth 在 dispatcher 心跳主题上应答探活，让管理端「服务状态」判定其在线。 +func (s *Subscriber) ServeHealth(provide func() []byte) (func() error, error) { + return s.inner.ServeHealth(contract.SubjectHealthDispatcher, provide) +} + // RequestModelConfig 向控制面（Gateway）取当前激活的对话模型配置。 func (s *Subscriber) RequestModelConfig(ctx context.Context) (*contract.ModelConfig, error) { return s.inner.RequestConfig(ctx, contract.ConfigKindChat) diff --git a/sundynix-gateway/internal/handler/status_handler.go b/sundynix-gateway/internal/handler/status_handler.go new file mode 100644 index 0000000..6c437a8 --- /dev/null +++ b/sundynix-gateway/internal/handler/status_handler.go @@ -0,0 +1,196 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "time" + + "github.com/gin-gonic/gin" + + "github.com/sundynix/sundynix-shared/contract" +) + +// statusItem 是一项依赖/服务的存活状态（基建灯、服务灯共用）。 +type statusItem struct { + Name string `json:"name"` + Up bool `json:"up"` + Detail string `json:"detail,omitempty"` + Latency int `json:"latency_ms,omitempty"` // NATS 探针往返耗时（毫秒），本地检查项为 0 +} + +// toolInfo 是一个注册工具的元信息（透传各 MCP 服务 list_tools 的上报）。 +type toolInfo struct { + Name string `json:"name"` + CN string `json:"cn"` + Desc string `json:"desc"` +} + +// toolGroup 是一台 MCP 服务的工具注册情况。 +type toolGroup struct { + Server string `json:"server"` + Up bool `json:"up"` + Tools []toolInfo `json:"tools"` +} + +// systemStatus 是「服务状态」面板的聚合视图：基建 / 应用服务 / MCP 工具注册。 +type systemStatus struct { + CheckedAt string `json:"checked_at"` + Infra []statusItem `json:"infra"` + Services []statusItem `json:"services"` + Tools []toolGroup `json:"tools"` +} + +// probeTimeout 是各探针的单次超时（无响应即判为下线）。 +const probeTimeout = 2 * time.Second + +// AdminStatus: GET /api/v1/admin/status —— 聚合基建、应用服务与 MCP 工具注册的实时状态， +// 供管理端「服务状态」一眼看出哪个服务没起、基建是否就绪、工具是否注册。 +func (h *Handler) AdminStatus(c *gin.Context) { + parent := c.Request.Context() + + // 各探针互不依赖，并发执行，整体只等最慢的一个。 + var ( + wg sync.WaitGroup + + milvus, neo4j bool // mcp-go health + goUp bool // mcp-go 在线 + goTools []toolInfo // mcp-go 注册工具 + goLatency int // mcp-go 探针耗时 + pyUp bool // mcp-py 在线 + pyTools []toolInfo // mcp-py 注册工具 + pyLatency int // mcp-py 探针耗时 + dispUp bool // dispatcher 在线 + dispDetail string // dispatcher 详情（模型/运行时长） + dispLatency int // dispatcher 探针耗时 + ) + + wg.Add(4) + + // 1) mcp-go health → milvus / neo4j 基建灯 + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + if res, err := h.bus.CallTool(ctx, contract.ToolSubjectGo("health"), + &contract.ToolCall{Tool: "health"}); err == nil && res != nil && res.OK { + var sub map[string]bool + if json.Unmarshal([]byte(res.Content), &sub) == nil { + milvus, neo4j = sub["milvus"], sub["neo4j"] + } + } + }() + + // 2) mcp-go list_tools → 在线判定 + 工具清单 + go func() { + defer wg.Done() + goUp, goTools, goLatency = h.probeTools(parent, contract.ToolSubjectGo("list_tools")) + }() + + // 3) mcp-py list_tools → 在线判定 + 工具清单 + go func() { + defer wg.Done() + pyUp, pyTools, pyLatency = h.probeTools(parent, contract.ToolSubjectPy("list_tools")) + }() + + // 4) dispatcher 心跳 → 在线判定 + 模型/运行时长 + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + start := time.Now() + if data, err := h.bus.Ping(ctx, contract.SubjectHealthDispatcher); err == nil { + dispUp = true + dispLatency = int(time.Since(start).Milliseconds()) + var st struct { + Model string `json:"model"` + Ready bool `json:"ready"` + UptimeS int `json:"uptime_s"` + } + if json.Unmarshal(data, &st) == nil { + dispDetail = dispatcherDetail(st.Model, st.Ready, st.UptimeS) + } + } + }() + + wg.Wait() + + c.JSON(http.StatusOK, systemStatus{ + CheckedAt: time.Now().Format(time.RFC3339), + Infra: []statusItem{ + {Name: "postgres", Up: h.db.Enabled()}, + {Name: "redis", Up: h.cache.Enabled()}, + {Name: "nats", Up: true}, // 网关连不上 NATS 即 fatal，能应答即在线 + {Name: "milvus", Up: milvus}, + {Name: "neo4j", Up: neo4j}, + }, + Services: []statusItem{ + {Name: "gateway", Up: true, Detail: "在线"}, + {Name: "dispatcher", Up: dispUp, Detail: serviceDetail(dispUp, dispDetail), Latency: dispLatency}, + {Name: "mcp-go", Up: goUp, Detail: toolsDetail(goUp, len(goTools)), Latency: goLatency}, + {Name: "mcp-py", Up: pyUp, Detail: toolsDetail(pyUp, len(pyTools)), Latency: pyLatency}, + }, + Tools: []toolGroup{ + {Server: "mcp-go", Up: goUp, Tools: goTools}, + {Server: "mcp-py", Up: pyUp, Tools: pyTools}, + }, + }) +} + +// probeTools 调一台 MCP 服务的 list_tools：能应答即在线，并解析其工具清单（含中文名/作用）+ 往返耗时。 +func (h *Handler) probeTools(parent context.Context, subject string) (up bool, tools []toolInfo, latency int) { + ctx, cancel := context.WithTimeout(parent, probeTimeout) + defer cancel() + start := time.Now() + res, err := h.bus.CallTool(ctx, subject, &contract.ToolCall{Tool: "list_tools"}) + if err != nil || res == nil || !res.OK { + return false, nil, 0 + } + var payload struct { + Tools []toolInfo `json:"tools"` + } + _ = json.Unmarshal([]byte(res.Content), &payload) + return true, payload.Tools, int(time.Since(start).Milliseconds()) +} + +func dispatcherDetail(model string, ready bool, uptimeS int) string { + d := "运行 " + humanDuration(uptimeS) + if model != "" { + d = "模型 " + model + " · " + d + } + if !ready { + d += "（模型未配置，降级桩）" + } + return d +} + +func serviceDetail(up bool, detail string) string { + if !up { + return "无响应（未启动？）" + } + if detail == "" { + return "在线" + } + return detail +} + +func toolsDetail(up bool, n int) string { + if !up { + return "无响应（未启动？）" + } + return fmt.Sprintf("%d 个工具", n) +} + +// humanDuration 把秒数转人读（< 1h 显示分钟，否则小时+分钟）。 +func humanDuration(s int) string { + if s < 60 { + return fmt.Sprintf("%ds", s) + } + m := s / 60 + if m < 60 { + return fmt.Sprintf("%dm", m) + } + return fmt.Sprintf("%dh%dm", m/60, m%60) +} diff --git a/sundynix-gateway/internal/nats/publisher.go b/sundynix-gateway/internal/nats/publisher.go index 6b97edc..cd328f8 100644 --- a/sundynix-gateway/internal/nats/publisher.go +++ b/sundynix-gateway/internal/nats/publisher.go @@ -54,6 +54,11 @@ func (b *Bus) CallTool(ctx context.Context, subject string, call *contract.ToolC return b.inner.CallTool(ctx, subject, call) } +// Ping 同步探测某节点健康（如 dispatcher 心跳主题）。无人应答 / 超时即返回错误（视为下线）。 +func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) { + return b.inner.Ping(ctx, subject) +} + // ServeConfig 让网关作为配置控制面，响应某 kind 的配置请求。 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (func() error, error) { return b.inner.ServeConfig(kind, provide) diff --git a/sundynix-gateway/internal/router/router.go b/sundynix-gateway/internal/router/router.go index 6f9ee16..0bb7275 100644 --- a/sundynix-gateway/internal/router/router.go +++ b/sundynix-gateway/internal/router/router.go @@ -79,6 +79,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob. admin.POST("/models/test", h.TestModel) admin.GET("/pricing", h.ListPricing) // 各模型计价（token↔真钱） admin.PUT("/pricing", h.SavePricing) // 设置某模型输入/输出单价 + admin.GET("/status", h.AdminStatus) // 服务状态：基建/服务探活 + MCP 工具注册 } } return r diff --git a/sundynix-mcp-go/internal/mcp/gateway.go b/sundynix-mcp-go/internal/mcp/gateway.go index 58edcc5..730db3b 100644 --- a/sundynix-mcp-go/internal/mcp/gateway.go +++ b/sundynix-mcp-go/internal/mcp/gateway.go @@ -8,6 +8,7 @@ import ( "log" "os" "path/filepath" + "sort" "strings" "time" @@ -28,10 +29,20 @@ type Gateway struct { memory *memory.Store history *history.Store rag *rag.Engine + tools map[string]toolDef // 工具注册表：唯一事实源，dispatch 与 list_tools 共用，杜绝漂移 +} + +// toolDef 是一个注册工具的元信息（中文名 / 作用）+ 处理函数。 +type toolDef struct { + cn string // 中文名 + desc string // 作用简述 + handler func(context.Context, *contract.ToolCall) *contract.ToolResult } func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store, r *rag.Engine) *Gateway { - return &Gateway{bus: b, search: s, memory: m, history: h, rag: r} + g := &Gateway{bus: b, search: s, memory: m, history: h, rag: r} + g.tools = g.buildRegistry() + return g } // Serve 以队列组通配订阅 sundynix.tools.go.>，按工具名分发并阻塞。 @@ -47,46 +58,64 @@ func (g *Gateway) Serve(ctx context.Context) error { return ctx.Err() } -// dispatch 按 ToolCall.Tool 路由到具体工具实现。 +// buildRegistry 注册 mcp-go 全部工具：名称 → (中文名, 作用, 处理函数)。 +// 这是工具的唯一事实源——dispatch 据此路由、list_tools 据此上报，二者永不漂移。 +func (g *Gateway) buildRegistry() map[string]toolDef { + return map[string]toolDef{ + "wiki_search": {"知识检索", "向量检索知识库（Milvus），返回最相关片段", g.wikiSearch}, + "kb_ingest": {"知识入库", "文本切块 → 向量化 → 写入 Milvus / Bleve", g.kbIngest}, + "kb_search": {"检索台查询", "结构化返回命中内容与相似度分数", g.kbSearch}, + "kb_graph": {"知识图谱", "取某库的实体关系三元组（Neo4j）", g.kbGraph}, + "report_render": {"报告渲染", "把结构化报告渲染为 Word（.docx）", g.reportRender}, + "report_store": {"报告存源", "暂存报告源数据，供导出时按需渲染", g.reportStore}, + "report_export": {"报告导出", "按需把已存报告导出为 Word / Markdown", g.reportExport}, + "external_api": {"外部接口", "受控调用第三方 HTTP API（带 SSRF 校验）", g.externalAPI}, + "memory_get": {"记忆召回", "取用户长期画像（已按打分排序）", g.memoryGet}, + "memory_upsert": {"记忆写入", "新增 / 更新一条用户偏好（带重要度）", g.memoryUpsert}, + "memory_delete": {"记忆删除", "软删一条偏好（对账判定过时 / 矛盾时）", g.memoryDelete}, + "memory_list": {"记忆列表", "列出用户全部偏好（供管理面板查看）", g.memoryList}, + "history_get": {"历史召回", "取会话最近多轮对话", g.historyGet}, + "history_append": {"历史追加", "往会话写入一条消息", g.historyAppend}, + "health": {"健康检查", "上报 Milvus / Neo4j / embedding 就绪情况", + func(_ context.Context, _ *contract.ToolCall) *contract.ToolResult { + data, _ := json.Marshal(g.rag.Status()) + return &contract.ToolResult{OK: true, Content: string(data)} + }}, + "echo": {"回显", "原样返回入参（调试用）", + func(_ context.Context, call *contract.ToolCall) *contract.ToolResult { + return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])} + }}, + } +} + +// dispatch 按 ToolCall.Tool 从注册表路由到具体工具实现。 +// list_tools 是元工具（自省），不在业务注册表内，单独处理。 func (g *Gateway) dispatch(ctx context.Context, call *contract.ToolCall) *contract.ToolResult { log.Printf("[mcp_go] tool=%s task=%s args=%v", call.Tool, call.TaskID, call.Args) - switch call.Tool { - case "wiki_search": - return g.wikiSearch(ctx, call) - case "kb_ingest": - return g.kbIngest(ctx, call) - case "kb_search": - return g.kbSearch(ctx, call) - case "kb_graph": - return g.kbGraph(ctx, call) - case "report_render": - return g.reportRender(ctx, call) - case "report_store": - return g.reportStore(ctx, call) - case "report_export": - return g.reportExport(ctx, call) - case "external_api": - return g.externalAPI(ctx, call) - case "health": - data, _ := json.Marshal(g.rag.Status()) - return &contract.ToolResult{OK: true, Content: string(data)} - case "memory_get": - return g.memoryGet(ctx, call) - case "memory_upsert": - return g.memoryUpsert(ctx, call) - case "memory_delete": - return g.memoryDelete(ctx, call) - case "memory_list": - return g.memoryList(ctx, call) - case "history_get": - return g.historyGet(ctx, call) - case "history_append": - return g.historyAppend(ctx, call) - case "echo": - return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])} - default: + if call.Tool == "list_tools" { + return g.listTools() + } + td, ok := g.tools[call.Tool] + if !ok { return &contract.ToolResult{OK: false, Error: "unknown tool: " + call.Tool} } + return td.handler(ctx, call) +} + +// listTools 自省：上报本服务注册的工具清单（名称 + 中文名 + 作用），供管理端展示。 +func (g *Gateway) listTools() *contract.ToolResult { + type info struct { + Name string `json:"name"` + CN string `json:"cn"` + Desc string `json:"desc"` + } + out := make([]info, 0, len(g.tools)) + for name, td := range g.tools { + out = append(out, info{Name: name, CN: td.cn, Desc: td.desc}) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) // map 无序 → 稳定输出 + data, _ := json.Marshal(map[string]any{"service": "mcp-go", "tools": out}) + return &contract.ToolResult{OK: true, Content: string(data)} } // memoryGet 召回某用户的常驻画像（已渲染为可注入 prompt 的多行文本）。 diff --git a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py index 251b2b3..4b60186 100644 --- a/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py +++ b/sundynix-mcp-py/src/sundynix_mcp_py/mcp_gateway.py @@ -25,6 +25,14 @@ log = logging.getLogger("mcp_py") SUBJECT_PY_ALL = "sundynix.tools.py.>" QUEUE_PY = "mcp-py-workers" +# 工具元信息：名称 → (中文名, 作用简述)。list_tools 据此上报给管理端展示。 +TOOL_META = { + "echo": ("回显", "原样返回入参（调试用）"), + "run_code": ("代码执行", "静态守卫 + Docker 隔离沙箱运行代码（标准档 256m/10s）"), + "parse_document": ("文档解析", "文件 → 纯文本（MinerU / PaddleOCR）"), + "secure_sandbox": ("安全沙箱", "更严资源档（128m/5s）的隔离执行，用于高风险代码"), +} + class McpGateway: def __init__(self) -> None: @@ -41,6 +49,7 @@ class McpGateway: "run_code": self._run_code, "parse_document": self._parse_document, "secure_sandbox": self._secure_sandbox, + "list_tools": self._list_tools, } async def serve(self, url: str | None = None) -> None: @@ -92,6 +101,15 @@ class McpGateway: async def _echo(self, args: dict) -> str: return str(args.get("text", "")) + async def _list_tools(self, args: dict) -> str: + """自省：上报业务工具清单（名称 + 中文名 + 作用），供管理端探活 + 展示。""" + tools = [ + {"name": n, "cn": cn, "desc": d} + for n, (cn, d) in TOOL_META.items() + if n in self._tools # 仅上报真正注册的业务工具（list_tools 自身不计入） + ] + return json.dumps({"service": "mcp-py", "tools": tools}) + async def _run_code(self, args: dict) -> str: """静态守卫 → Docker 隔离执行（标准档 256m/0.5cpu/10s）。""" code = str(args.get("code", "")) diff --git a/sundynix-shared/bus/bus.go b/sundynix-shared/bus/bus.go index c18e44b..63631cf 100644 --- a/sundynix-shared/bus/bus.go +++ b/sundynix-shared/bus/bus.go @@ -214,6 +214,29 @@ func respond(m *nats.Msg, res *contract.ToolResult) { _ = m.Respond(data) } +// ---- 服务探活（core NATS request-reply 心跳）---- + +// ServeHealth 在 subject 上应答健康探测，provide 返回本节点状态 JSON（可为空）。 +// 用于无 HTTP/工具端点的节点（如 dispatcher）向控制面暴露存活。 +func (b *Bus) ServeHealth(subject string, provide func() []byte) (unsub func() error, err error) { + sub, err := b.nc.Subscribe(subject, func(m *nats.Msg) { + _ = m.Respond(provide()) + }) + if err != nil { + return nil, fmt.Errorf("serve health %s: %w", subject, err) + } + return sub.Unsubscribe, nil +} + +// Ping 同步探测某节点健康：发到 subject 等应答。无人应答 / 超时即返回错误（视为下线）。 +func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) { + msg, err := b.nc.RequestWithContext(ctx, subject, nil) + if err != nil { + return nil, err + } + return msg.Data, nil +} + // ---- 配置控制面（core NATS request-reply + broadcast）---- // RequestConfig 向控制面（Gateway）请求某 kind 当前激活配置（chat/embedding）。 diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index ad0cd1a..73c7614 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -25,6 +25,10 @@ const ( QueueToolsGo = "mcp-go-workers" // mcp-go 队列组（多副本负载均衡） QueueToolsPy = "mcp-py-workers" // mcp-py 队列组 + // 服务探活：dispatcher 既无 HTTP 端点也不挂工具，单独用一个 core NATS + // request-reply 心跳主题让控制面（管理端「服务状态」）能判定它在不在线。 + SubjectHealthDispatcher = "sundynix.health.dispatcher" + // MetaUserID 是 Task.Meta 中承载已登录用户标识的键（用于偏好记忆召回）。 MetaUserID = "user_id" // MetaSessionID 是 Task.Meta 中承载会话标识的键（用于短期多轮历史）。