feat: 打通 Dispatcher→MCP 工具调用链路 (core NATS request-reply)

第 4 层 Dispatcher 经 NATS request-reply + 队列组同步调用第 5 层 MCP 工具，工具不可用/超时即降级，不阻断主流程。 - shared/contract: ToolCall/ToolResult + sundynix.tools.go.* subject 约定 + ToolSubjectGo/Py - shared/bus: CallTool(发起) / ServeTool(队列组订阅+应答) - mcp-go: 接共享 bus，gateway 通配订阅按工具名分发(wiki_search/echo)，main 优雅退出 - dispatcher: ToolCaller 接口 + Orchestrator.retrieveContext(调 wiki_search,超时3s降级) - e2e: TestToolCallRoundTrip(PASS)；demo.sh 加 mcp-go(就绪门避免启动竞态)，live 跑通 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 11:31:58 +08:00
parent 61337b1920
commit adc521f94d
12 changed files with 315 additions and 32 deletions
@@ -19,7 +19,7 @@

 1. `Gateway` 解析 DSL → **Publish** `sundynix.tasks.*` (NATS Queue)
 2. `Dispatcher` 订阅任务 → `Eino` 图编排 → 调用 `LLM Pool`
-3. 经 NATS 调用第 5 层 `MCP Tools`
+3. 经 NATS **request-reply** 同步调用第 5 层 `MCP Tools`（`sundynix.tools.go.*` 队列组负载均衡，工具不可用即降级）
 4. 结果以零拷贝 Token Stream 经 `sundynix.streams.task_id` 回流 → SSE/WS 推给 `Client`

 ## 快速开始
@@ -27,8 +27,8 @@
 ### 无 Docker — 一键验证任务流（推荐先跑这个）

 ```bash
-make demo   # 内嵌NATS + Gateway + Dispatcher，提交一个 DSL 任务，看 Dispatcher 消费到
-make e2e    # 仅跑共享 bus 的端到端测试（go test，内嵌 NATS）
+make demo   # 内嵌NATS + Gateway + Dispatcher + MCP-Go，提交 DSL 任务，看任务流+工具调用+Token流闭环
+make e2e    # 仅跑共享 bus 的端到端测试（go test，内嵌 NATS）：任务流 / 工具调用 / Token 流
 ```

 `make demo` 实测输出：
@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
-# 无 Docker 的最小任务流演示：devnats(内嵌NATS) + gateway + dispatcher。
-# 提交一个 DSL 任务，验证 Gateway → NATS → Dispatcher 全链路。
+# 无 Docker 的最小任务流演示：devnats(内嵌NATS) + gateway + dispatcher + mcp-go。
+# 提交一个 DSL 任务，验证 Gateway → NATS → Dispatcher → MCP 工具 全链路。
 set -euo pipefail
 cd "$(dirname "$0")/.."

@@ -9,8 +9,9 @@ echo "== 编译 =="
 ( cd sundynix-shared     && go build -o ../.bin/devnats    ./cmd/devnats )
 ( cd sundynix-gateway    && go build -o ../.bin/gateway    ./cmd/server  )
 ( cd sundynix-dispatcher && go build -o ../.bin/dispatcher ./cmd/dispatcher )
+( cd sundynix-mcp-go     && go build -o ../.bin/mcp-go     ./cmd/server  )

-cleanup() { kill "${GW_PID:-}" "${DISP_PID:-}" "${NATS_PID:-}" 2>/dev/null || true; }
+cleanup() { kill "${GW_PID:-}" "${DISP_PID:-}" "${MCP_PID:-}" "${NATS_PID:-}" 2>/dev/null || true; }
 trap cleanup EXIT

 # 若 :4222 已有 NATS（docker compose 的容器），直接复用；否则起内嵌 devnats。
@@ -22,7 +23,8 @@ else
  for _ in $(seq 1 30); do nc -z 127.0.0.1 4222 2>/dev/null && break || sleep 0.2; done
 fi

-echo "== 启动 dispatcher / gateway =="
+echo "== 启动 mcp-go / dispatcher / gateway =="
+.bin/mcp-go     > .bin/mcp-go.log      2>&1 & MCP_PID=$!
 .bin/dispatcher > .bin/dispatcher.log  2>&1 & DISP_PID=$!
 .bin/gateway    > .bin/gateway.log     2>&1 & GW_PID=$!

@@ -30,6 +32,11 @@ for _ in $(seq 1 30); do
  curl -s -o /dev/null http://127.0.0.1:8080/api/v1/billing && break || sleep 0.3
 done

+# 等 mcp-go 订阅就绪后再提交，否则工具调用会撞上启动竞态而降级。
+for _ in $(seq 1 30); do
+  grep -q "tools ready" .bin/mcp-go.log 2>/dev/null && break || sleep 0.1
+done
+
 echo "== 提交 DSL 任务 =="
 RESP=$(curl -s -X POST http://127.0.0.1:8080/api/v1/tasks \
  -H 'Content-Type: application/json' \
@@ -42,5 +49,8 @@ echo "== 订阅 SSE Token 流 (Gateway ← NATS ← Dispatcher) =="
 curl -sN --max-time 10 "http://127.0.0.1:8080/api/v1/tasks/$TASK_ID/stream" || true
 echo

+echo "== mcp-go 日志 (工具被调用) =="
+cat .bin/mcp-go.log
+
 echo "== dispatcher 日志 =="
 cat .bin/dispatcher.log
@@ -23,8 +23,8 @@ func main() {
 	sub := dnats.MustConnect(natsURL)
 	defer sub.Close()

-	// sub 同时作为 Token 回流出口（TokenSink）。
-	orch := eino.NewOrchestrator(pool, breaker, sub)
+	// sub 同时作为 Token 回流出口（TokenSink）与 MCP 工具调用出口（ToolCaller）。
+	orch := eino.NewOrchestrator(pool, breaker, sub, sub)

 	// 监听退出信号，优雅停止消费。
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -4,6 +4,7 @@ package eino
 import (
 	"context"
 	"log"
+	"time"

 	"github.com/sundynix/sundynix-dispatcher/internal/harness"
 	"github.com/sundynix/sundynix-dispatcher/internal/llm"
@@ -16,15 +17,24 @@ type TokenSink interface {
 	CompleteStream(taskID string) error
 }

+// ToolCaller 经 NATS 调起第 5 层 MCP 工具（由 NATS bus 实现）。
+type ToolCaller interface {
+	CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error)
+}
+
+// 工具调用超时；超时即降级（不带工具上下文继续推理）。
+const toolCallTimeout = 3 * time.Second
+
 // Orchestrator 将 DSL 图编译为 Eino Graph 并驱动执行。
 type Orchestrator struct {
 	pool    *llm.Pool
 	breaker *harness.CircuitBreaker
 	sink    TokenSink
+	tools   ToolCaller
 }

-func NewOrchestrator(pool *llm.Pool, breaker *harness.CircuitBreaker, sink TokenSink) *Orchestrator {
-	return &Orchestrator{pool: pool, breaker: breaker, sink: sink}
+func NewOrchestrator(pool *llm.Pool, breaker *harness.CircuitBreaker, sink TokenSink, tools ToolCaller) *Orchestrator {
+	return &Orchestrator{pool: pool, breaker: breaker, sink: sink, tools: tools}
 }

 // Handle 消费一个任务：编译图 → 流式推理 → 经 sink 把 Token 回流到 sundynix.streams.<id>。
@@ -36,9 +46,14 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
 	log.Printf("[eino] task %s received (graph=%d bytes), streaming tokens...", t.ID, len(t.Graph))

 	// TODO: compose.NewGraph(...) 编译 DSL；此处 prompt 占位为图原文。
-	//       工具节点经 NATS 调用第 5 层 MCP（sundynix.tools.go.* / sundynix.tools.py.*）。
 	prompt := string(t.Graph)

+	// 工具节点：经 NATS 调用第 5 层 MCP（sundynix.tools.go.*）。
+	// 这里以 wiki_search 演示完整调用链路；真实 Eino 图会按 DSL 节点择机调用。
+	if ctxNote := o.retrieveContext(ctx, t); ctxNote != "" {
+		prompt = ctxNote + "\n" + prompt
+	}
+
 	n := 0
 	err := o.pool.Stream(ctx, prompt, func(tok []byte) {
 		if perr := o.sink.PublishToken(t.ID, tok); perr != nil {
@@ -58,3 +73,29 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
 	o.breaker.Report(err == nil)
 	return err
 }
+
+// retrieveContext 经 MCP wiki_search 工具拉取检索上下文。
+// 工具不可用/超时时返回空串，降级为无工具上下文推理（不阻断主流程）。
+func (o *Orchestrator) retrieveContext(ctx context.Context, t *contract.Task) string {
+	if o.tools == nil {
+		return ""
+	}
+	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
+	defer cancel()
+
+	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("wiki_search"), &contract.ToolCall{
+		Tool:   "wiki_search",
+		TaskID: t.ID,
+		Args:   map[string]any{"q": string(t.Graph)},
+	})
+	if err != nil {
+		log.Printf("[eino] task %s wiki_search unavailable, degrade: %v", t.ID, err)
+		return ""
+	}
+	if !res.OK {
+		log.Printf("[eino] task %s wiki_search error: %s", t.ID, res.Error)
+		return ""
+	}
+	log.Printf("[eino] task %s wiki_search ok: %s", t.ID, res.Content)
+	return res.Content
+}
@@ -53,4 +53,9 @@ func (s *Subscriber) CompleteStream(taskID string) error {
 	return s.inner.CompleteStream(taskID)
 }

+// CallTool 让 Subscriber 满足 eino.ToolCaller，经 NATS request-reply 调起第 5 层 MCP 工具。
+func (s *Subscriber) CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error) {
+	return s.inner.CallTool(ctx, subject, call)
+}
+
 func (s *Subscriber) Close() { s.inner.Close() }
@@ -2,18 +2,43 @@
 package main

 import (
+	"context"
 	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	sharedbus "github.com/sundynix/sundynix-shared/bus"

 	"github.com/sundynix/sundynix-mcp-go/internal/mcp"
 	"github.com/sundynix/sundynix-mcp-go/internal/search"
 )

 func main() {
-	engine := search.NewHybrid() // LLM Wiki 混合检索：Bleve + Milvus + Neo4j
-	gw := mcp.NewGateway(engine)
+	natsURL := envOr("NATS_URL", "nats://localhost:4222")

-	log.Println("[mcp_go] serving MCP over sundynix.tools.go.*")
-	if err := gw.Serve(); err != nil {
+	b, err := sharedbus.Connect(natsURL)
+	if err != nil {
+		log.Fatalf("[mcp_go] nats connect: %v", err)
+	}
+	defer b.Close()
+	log.Printf("[mcp_go] connected %s", natsURL)
+
+	engine := search.NewHybrid() // LLM Wiki 混合检索：Bleve + Milvus + Neo4j
+	gw := mcp.NewGateway(b, engine)
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	log.Println("[mcp_go] serving MCP over sundynix.tools.go.* (Ctrl-C to quit)")
+	if err := gw.Serve(ctx); err != nil && err != context.Canceled {
 		log.Fatalf("[mcp_go] exit: %v", err)
 	}
 }
+
+func envOr(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
@@ -2,9 +2,16 @@ module github.com/sundynix/sundynix-mcp-go

 go 1.23

+require github.com/sundynix/sundynix-shared v0.0.0
+
 require (
-	github.com/blevesearch/bleve/v2 v2.4.2
-	github.com/milvus-io/milvus-sdk-go/v2 v2.4.1
-	github.com/neo4j/neo4j-go-driver/v5 v5.24.0
-	github.com/nats-io/nats.go v1.37.0
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/nats-io/nats.go v1.37.0 // indirect
+	github.com/nats-io/nkeys v0.4.7 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	golang.org/x/crypto v0.26.0 // indirect
+	golang.org/x/sys v0.24.0 // indirect
+	golang.org/x/text v0.17.0 // indirect
 )
+
+replace github.com/sundynix/sundynix-shared => ../sundynix-shared
@@ -1,4 +1,22 @@
-github.com/blevesearch/bleve/v2 v2.4.2/go.mod h1:ATNKj7Yl2oJv/lGuF4kx39bST2dveX6w0th2FFYLkc8=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
+github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.5.8 h1:uvdSzwWiEGWGXf+0Q+70qv6AQdvcvxrv9hPM0RiPamE=
+github.com/nats-io/jwt/v2 v2.5.8/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
+github.com/nats-io/nats-server/v2 v2.10.20 h1:CXDTYNHeBiAKBTAIP2gjpgbWap2GhATnTLgP8etyvEI=
+github.com/nats-io/nats-server/v2 v2.10.20/go.mod h1:hgcPnoUtMfxz1qVOvLZGurVypQ+Cg6GXVXjG53iHk+M=
+github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE=
 github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
-github.com/neo4j/neo4j-go-driver/v5 v5.24.0/go.mod h1:Vff8OwT7QpLm7L2yYr85XNWe9Rbqlbeb9asNXJTHO4k=
-github.com/qdrant/go-client v1.11.0/go.mod h1:j+OVRsJIZhOSRK2toPl8tTBOhwr4AxXCz9RACzv0JB4=
+github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
+github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
+golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
+golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
+golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
+golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/time v0.6.0 h1:eTDhh4ZXt5Qf0augr54TN6suAUudPcawVZeIAPU7D4U=
+golang.org/x/time v0.6.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
@@ -1,18 +1,71 @@
 // Package mcp 实现 MCP 协议网关，把工具注册到 NATS 并响应调用。
 package mcp

-import "github.com/sundynix/sundynix-mcp-go/internal/search"
+import (
+	"context"
+	"fmt"
+	"log"

-// Gateway 暴露 MCP 协议端点（stdio / HTTP / NATS）。
+	sharedbus "github.com/sundynix/sundynix-shared/bus"
+	"github.com/sundynix/sundynix-shared/contract"
+
+	"github.com/sundynix/sundynix-mcp-go/internal/search"
+)
+
+// Gateway 暴露 MCP 协议端点，经共享 bus 订阅 sundynix.tools.go.* 响应调用。
 type Gateway struct {
+	bus    *sharedbus.Bus
 	search *search.Hybrid
 }

-func NewGateway(s *search.Hybrid) *Gateway { return &Gateway{search: s} }
-
-// Serve 监听 sundynix.tools.go.* 并按 MCP 协议分发工具调用。
-func (g *Gateway) Serve() error {
-	// TODO: 注册工具清单 (wiki_search / render_doc / call_external_api ...)
-	//       订阅 NATS，按 MCP JSON-RPC 解析并路由
-	select {}
+func NewGateway(b *sharedbus.Bus, s *search.Hybrid) *Gateway {
+	return &Gateway{bus: b, search: s}
+}
+
+// Serve 以队列组通配订阅 sundynix.tools.go.>，按工具名分发并阻塞。
+func (g *Gateway) Serve(ctx context.Context) error {
+	unsub, err := g.bus.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo, g.dispatch)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = unsub() }()
+	log.Printf("[mcp_go] tools ready on %s (queue=%s): wiki_search, echo",
+		contract.SubjectToolsGoAll, contract.QueueToolsGo)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// dispatch 按 ToolCall.Tool 路由到具体工具实现。
+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 "echo":
+		return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])}
+	default:
+		return &contract.ToolResult{OK: false, Error: "unknown tool: " + call.Tool}
+	}
+}
+
+// wikiSearch 调 Hybrid 混合检索引擎。引擎目前为桩（返回空），
+// 这里仍把调用链路做真：真实接入 Bleve/Milvus/Neo4j 后无需改动协议。
+func (g *Gateway) wikiSearch(ctx context.Context, call *contract.ToolCall) *contract.ToolResult {
+	q, _ := call.Args["q"].(string)
+	results, err := g.search.Query(ctx, q, 5)
+	if err != nil {
+		return &contract.ToolResult{OK: false, Error: "wiki_search: " + err.Error()}
+	}
+	return &contract.ToolResult{
+		OK:      true,
+		Content: fmt.Sprintf("[wiki_search] 命中 %d 条（Bleve+Milvus+Neo4j 混合检索桩）查询=%q", len(results), preview(q)),
+	}
+}
+
+func preview(s string) string {
+	r := []rune(s)
+	if len(r) > 40 {
+		return string(r[:40]) + "…"
+	}
+	return s
 }
@@ -4,6 +4,7 @@ package bus

 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"

@@ -133,6 +134,55 @@ func (b *Bus) SubscribeTokens(taskID string, onToken func([]byte), onDone func()
 	return sub.Unsubscribe, nil
 }

+// ---- MCP 工具调用（core NATS request-reply）----
+
+// CallTool 同步调用一个 MCP 工具：发到 subject，阻塞等待应答。
+// ctx 超时即视为工具不可用，由调用方决定降级。
+func (b *Bus) CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error) {
+	data, err := json.Marshal(call)
+	if err != nil {
+		return nil, fmt.Errorf("marshal tool call: %w", err)
+	}
+	msg, err := b.nc.RequestWithContext(ctx, subject, data)
+	if err != nil {
+		return nil, fmt.Errorf("call tool %s: %w", subject, err)
+	}
+	var res contract.ToolResult
+	if err := json.Unmarshal(msg.Data, &res); err != nil {
+		return nil, fmt.Errorf("unmarshal tool result: %w", err)
+	}
+	return &res, nil
+}
+
+// ToolHandler 处理一次工具调用并返回结果。
+type ToolHandler func(ctx context.Context, call *contract.ToolCall) *contract.ToolResult
+
+// ServeTool 以队列组订阅工具主题（可用通配 sundynix.tools.go.>），
+// 对每个请求调用 h 并 Respond，队列组内多副本自动负载均衡。
+// 返回的 unsub 用于退订。
+func (b *Bus) ServeTool(subject, queue string, h ToolHandler) (unsub func() error, err error) {
+	sub, err := b.nc.QueueSubscribe(subject, queue, func(m *nats.Msg) {
+		var call contract.ToolCall
+		if err := json.Unmarshal(m.Data, &call); err != nil {
+			respond(m, &contract.ToolResult{OK: false, Error: "bad tool call: " + err.Error()})
+			return
+		}
+		respond(m, h(context.Background(), &call))
+	})
+	if err != nil {
+		return nil, fmt.Errorf("serve tool %s: %w", subject, err)
+	}
+	return sub.Unsubscribe, nil
+}
+
+func respond(m *nats.Msg, res *contract.ToolResult) {
+	data, err := json.Marshal(res)
+	if err != nil {
+		data, _ = json.Marshal(&contract.ToolResult{OK: false, Error: "marshal result: " + err.Error()})
+	}
+	_ = m.Respond(data)
+}
+
 // TaskHandler 处理一个消费到的任务。
 type TaskHandler func(ctx context.Context, t *contract.Task) error

@@ -93,6 +93,53 @@ func TestTaskRoundTrip(t *testing.T) {
 	}
 }

+// TestToolCallRoundTrip 模拟 Dispatcher 经 NATS 调用 → mcp-go 响应 的工具调用闭环。
+func TestToolCallRoundTrip(t *testing.T) {
+	url := startEmbeddedNATS(t)
+
+	// --- mcp-go 侧：以队列组订阅工具主题并响应 ---
+	srv, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("mcp connect: %v", err)
+	}
+	defer srv.Close()
+
+	unsub, err := srv.ServeTool(contract.SubjectToolsGoAll, contract.QueueToolsGo,
+		func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
+			if call.Tool != "wiki_search" {
+				return &contract.ToolResult{OK: false, Error: "unknown tool"}
+			}
+			return &contract.ToolResult{OK: true, Content: "命中:" + call.Args["q"].(string)}
+		})
+	if err != nil {
+		t.Fatalf("serve tool: %v", err)
+	}
+	defer func() { _ = unsub() }()
+
+	// --- Dispatcher 侧：同步调用工具 ---
+	dp, err := bus.Connect(url)
+	if err != nil {
+		t.Fatalf("dispatcher connect: %v", err)
+	}
+	defer dp.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	res, err := dp.CallTool(ctx, contract.ToolSubjectGo("wiki_search"), &contract.ToolCall{
+		Tool:   "wiki_search",
+		TaskID: "task_tool_001",
+		Args:   map[string]any{"q": "向量检索"},
+	})
+	if err != nil {
+		t.Fatalf("call tool: %v", err)
+	}
+	if !res.OK || res.Content != "命中:向量检索" {
+		t.Fatalf("tool result = %+v, want ok content=命中:向量检索", res)
+	}
+	t.Logf("✓ 工具调用闭环：Dispatcher → sundynix.tools.go.wiki_search → mcp-go → %q", res.Content)
+}
+
 // TestTokenStreamRoundTrip 模拟 Dispatcher 回流 Token → Gateway 订阅 的流式闭环。
 func TestTokenStreamRoundTrip(t *testing.T) {
 	url := startEmbeddedNATS(t)
@@ -15,6 +15,15 @@ const (
 	// HeaderStreamEnd 是 Token 流的结束信号（core NATS 消息头）。
 	// 置为 "1" 的消息体为空，表示该 task 的 Token 流结束。
 	HeaderStreamEnd = "X-Stream-End"
+
+	// MCP 工具调用约定（第 4 层 Dispatcher → 第 5 层 MCP Tools）。
+	// 用 core NATS request-reply：同步拿结果，队列组内负载均衡。
+	SubjectToolsGo    = "sundynix.tools.go"   // Go I/O 型工具前缀；实际 sundynix.tools.go.<tool>
+	SubjectToolsGoAll = "sundynix.tools.go.>" // mcp-go 通配订阅
+	SubjectToolsPy    = "sundynix.tools.py"   // Python 算法型工具前缀；实际 sundynix.tools.py.<tool>
+	SubjectToolsPyAll = "sundynix.tools.py.>" // mcp-py 通配订阅
+	QueueToolsGo      = "mcp-go-workers"      // mcp-go 队列组（多副本负载均衡）
+	QueueToolsPy      = "mcp-py-workers"      // mcp-py 队列组
 )

 // Task 是 DSL 解析组装后的可调度任务，在 NATS 上以 JSON 传输。
@@ -30,6 +39,24 @@ func TaskSubject(id string) string { return SubjectTasks + "." + id }
 // StreamSubject 返回某任务的 Token 回流主题。
 func StreamSubject(id string) string { return SubjectStream + "." + id }

+// ToolSubjectGo / ToolSubjectPy 返回某工具的调用主题。
+func ToolSubjectGo(tool string) string { return SubjectToolsGo + "." + tool }
+func ToolSubjectPy(tool string) string { return SubjectToolsPy + "." + tool }
+
+// ToolCall 是 Dispatcher 对一个 MCP 工具的调用请求（NATS request 体）。
+type ToolCall struct {
+	Tool   string         `json:"tool"`              // 工具名，如 wiki_search
+	Args   map[string]any `json:"args,omitempty"`    // 工具参数
+	TaskID string         `json:"task_id,omitempty"` // 触发该调用的任务（便于追踪）
+}
+
+// ToolResult 是 MCP 工具的应答（NATS reply 体）。
+type ToolResult struct {
+	OK      bool   `json:"ok"`
+	Content string `json:"content,omitempty"` // 工具产出（如检索结果文本）
+	Error   string `json:"error,omitempty"`   // 非空表示工具内部出错
+}
+
 // Marshal / Unmarshal 便捷方法。
 func (t *Task) Marshal() ([]byte, error)       { return json.Marshal(t) }
 func Unmarshal(b []byte) (*Task, error) {