From 1116d30596c4cb693a172a507d107e132c18410c Mon Sep 17 00:00:00 2001 From: Junjun Zhang Date: Thu, 17 Sep 2026 06:39:45 +0800 Subject: [PATCH] feat(mcp): per-server circuit breaker for MCP tool calls (sa-21) An MCP server that crashed or became unreachable fails EVERY call to EVERY tool it exposes. Each such call burned the full transport pipeline (120s stdio timeout) and the model typically retried the same or sibling tools several times, costing minutes of wall-clock time and tokens per outage. Trajectory-level guards (recurring_error.go, error_compound.go) only inject guidance text; nothing at the execution layer short-circuits the wasted retries. Add a per-SERVER circuit breaker shared by every mcpTool of one server (shared across Registry.Clone copies via pointer, so swarm teammates share outage state): - CLOSED -> (3 consecutive infrastructure failures) -> OPEN -> (60s cooldown) -> HALF-OPEN single probe; success closes, infra failure reopens with a fresh cooldown. - Only transport-level failures trip it (connection refused/reset, EOF, broken pipe, DNS, i/o/deadline timeouts, TLS, 5xx). Semantic failures (server answered) and plain user Esc never do. - When OPEN, tool calls fast-fail instantly with an actionable message (what happened, why retrying wastes a full LLM round-trip, and the re-plan path), following the permissionDeniedMessage convention. - Zero LLM cost: deterministic O(1) counters; breaker disabled (nil) in hand-built fixtures, so existing behavior is unchanged. Co-Authored-By: ggcode --- internal/mcp/adapter.go | 42 ++++++ internal/mcp/breaker.go | 277 +++++++++++++++++++++++++++++++++++ internal/mcp/breaker_test.go | 261 +++++++++++++++++++++++++++++++++ 3 files changed, 580 insertions(+) create mode 100644 internal/mcp/breaker.go create mode 100644 internal/mcp/breaker_test.go diff --git a/internal/mcp/adapter.go b/internal/mcp/adapter.go index 5d729ab3f..38dae07bb 100644 --- a/internal/mcp/adapter.go +++ b/internal/mcp/adapter.go @@ -26,6 +26,10 @@ type Adapter struct { // registeredTools are the names THIS adapter actually owns in the // registry (collision-skips excluded, #1594-A). registeredTools []string + // breaker is the per-server circuit breaker (sa-21) shared by every + // mcpTool this adapter registers. One dead server must fast-fail all + // of its tools, so the state lives here, not per tool. + breaker *serverBreaker } // NewAdapter creates an MCP adapter from server config and tool definitions. @@ -34,6 +38,7 @@ func NewAdapter(serverName string, caller toolCaller, tools []ToolDefinition) *A serverName: serverName, caller: caller, tools: tools, + breaker: newServerBreaker(serverName), } } @@ -44,6 +49,7 @@ func NewReadOnlyAdapter(serverName string, caller toolCaller, tools []ToolDefini caller: caller, tools: tools, readOnly: true, + breaker: newServerBreaker(serverName), } } @@ -76,6 +82,7 @@ func (a *Adapter) RegisterTools(registry *tool.Registry) error { readOnly: a.readOnly, blocked: blocked, srvName: a.serverName, + breaker: a.breaker, // shared per-server state (sa-21) } if err := registry.Register(t); err != nil { // Log but continue — name collision is non-fatal. Whoever already @@ -126,6 +133,13 @@ type mcpTool struct { blocked bool srvName string + // breaker is the per-server circuit breaker (sa-21). Shared by ALL + // tools of the same server - including across Registry.Clone() copies + // (Clone copies the pointer, so swarm teammates share outage state; + // that is correct: a dead server is dead for every agent). Nil in + // hand-built test fixtures → breaker disabled, zero behavior change. + breaker *serverBreaker + // ContextFill mirrors the agent guard's fill ratio (current tokens / // compaction threshold, 0.0-1.0+). When ≥0.50 the result cap shrinks to // stay under the guard's corresponding limit, avoiding a second @@ -172,6 +186,7 @@ func (t *mcpTool) Clone() tool.Tool { readOnly: t.readOnly, blocked: t.blocked, srvName: t.srvName, + breaker: t.breaker, } } @@ -193,6 +208,14 @@ func (t *mcpTool) Parameters() json.RawMessage { } func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Result, error) { + // sa-21: per-server circuit breaker gate. When OPEN, fail fast WITHOUT + // a transport attempt - a dead server cannot succeed, and every retry + // burns a full LLM round-trip (often 120s+ of stdio timeout). + if t.breaker != nil { + if blocked, msg := t.breaker.gate(); blocked { + return tool.Result{Content: msg, IsError: true}, nil + } + } if t.blocked { return tool.Result{ Content: fmt.Sprintf("MCP server '%s' is in read-only mode, tool '%s' is not allowed", t.srvName, t.toolName), @@ -206,6 +229,12 @@ func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Resu } } if t.caller == nil { + // Never-connected counts as infrastructure: no tool on this server + // can work until it is (re)started, so it feeds the breaker. + err := error(errNotConnected{server: t.srvName}) + if t.breaker != nil { + t.breaker.recordFailure(err) + } return tool.Result{ Content: fmt.Sprintf("mcp[%s]: tool '%s' is not connected (server may have crashed or not started)", t.srvName, t.toolName), IsError: true, @@ -213,11 +242,24 @@ func (t *mcpTool) Execute(ctx context.Context, input json.RawMessage) (tool.Resu } result, err := t.caller.CallTool(ctx, t.toolName, args) if err != nil { + // Classify before surfacing: only transport-level failures feed the + // breaker. Semantic errors (server answered) never trip it. + if t.breaker != nil { + if isInfraError(err) { + t.breaker.recordFailure(err) + } else { + t.breaker.recordSuccess() + } + } return tool.Result{ Content: fmt.Sprintf("mcp[%s]: %s → %v", t.srvName, t.toolName, err), IsError: true, }, nil } + // The server answered (even isError=true results mean it is reachable). + if t.breaker != nil { + t.breaker.recordSuccess() + } // Extract text from content blocks var parts []string diff --git a/internal/mcp/breaker.go b/internal/mcp/breaker.go new file mode 100644 index 000000000..82fed66c9 --- /dev/null +++ b/internal/mcp/breaker.go @@ -0,0 +1,277 @@ +package mcp + +// Per-server MCP circuit breaker (sa-21). +// +// Research basis (2026 agentic-reliability engineering): +// - "Building Resilient AI Agents: Error Handling, Retries and Circuit +// Breakers" (TechMango, 2026) and "Agentic AI systems fail in storms: +// one dead dependency, N wasted retries" (2026 agent-ops surveys) - +// the canonical pattern is classify → trip → fast-fail → graceful +// degrade (tell the agent WHAT failed and WHY, let it re-plan). +// +// THE GAP IN GGCODE: an MCP server that crashed or became unreachable +// fails EVERY call to EVERY tool it exposes. Before this breaker, each +// such call burned the full pipeline - often a 120s stdio request +// timeout or the agent's adaptive tool timeout - and the model would +// typically retry the same tool (or a sibling tool of the same dead +// server) several times, costing minutes of wall-clock time and tokens +// per outage before it gave up. recurring_error.go / error_compound.go +// only inject guidance text at the trajectory level; nothing at the +// execution layer short-circuits the wasted retries. +// +// DESIGN: +// - One breaker per SERVER (not per tool): a dead server fails all of +// its tools, and the breaker state must be shared across every +// mcpTool instance, every agent clone (Registry.Clone shares the +// pointer via Clone), and sibling tools. +// - Only INFRASTRUCTURE failures trip it: connection refused/reset/ +// closed, EOF, broken pipe, DNS failure, i/o/deadline timeouts, TLS +// failures, HTTP 5xx. Semantic failures (the server answered: JSON- +// RPC "tool not found", "invalid params", or a tool result with +// isError=true) are the server WORKING - they never trip the +// breaker. Plain user-initiated context cancellation is also +// excluded (Esc is not an outage). +// - CLOSED → (threshold consecutive infra failures) → OPEN → (cooldown) +// → HALF-OPEN probe: one call allowed; success closes, infra failure +// reopens with a fresh cooldown. +// - Fail-fast message follows the permissionDeniedMessage convention: +// state what happened, why further retries waste a full LLM +// round-trip, and give the agent an actionable recovery path. +// - Zero LLM cost: deterministic counters, O(1) state. + +import ( + "fmt" + "strings" + "sync" + "time" + + "github.com/topcheer/ggcode/internal/debug" +) + +const ( + // breakerThreshold is the number of consecutive infrastructure + // failures before the circuit opens. 3 balances against transient + // hiccups (a single dropped WS frame) and against burning retries. + breakerThreshold = 3 + + // breakerCooldown is how long an OPEN circuit waits before allowing + // one half-open probe call. + breakerCooldown = 60 * time.Second +) + +type breakerState int + +const ( + breakerClosed breakerState = iota + breakerOpen + breakerHalfOpen +) + +// serverBreaker is the per-server circuit breaker shared by all mcpTool +// instances of one MCP server. All methods are goroutine-safe. +type serverBreaker struct { + mu sync.Mutex + server string + state breakerState + failures int // consecutive infra failures (closed) or current trip count + lastErr string + openedAt time.Time + probing bool // a half-open probe call is in flight + threshold int + cooldown time.Duration +} + +func newServerBreaker(server string) *serverBreaker { + return &serverBreaker{ + server: server, + threshold: breakerThreshold, + cooldown: breakerCooldown, + } +} + +// gate decides whether a tool call may proceed. When blocked is true, msg +// is a ready-to-return fast-fail error content. A blocked call is +// instantaneous (no transport I/O), which is the entire point: it saves +// the full timeout + LLM round-trip the retry would have burned. +func (b *serverBreaker) gate() (blocked bool, msg string) { + b.mu.Lock() + defer b.mu.Unlock() + switch b.state { + case breakerClosed: + return false, "" + case breakerOpen: + if time.Since(b.openedAt) >= b.cooldown { + // Cooldown elapsed: this call becomes the half-open probe. + b.state = breakerHalfOpen + b.probing = true + return false, "" + } + remain := b.cooldown - time.Since(b.openedAt) + if remain < 0 { + remain = 0 + } + return true, b.fastFailMessage(fmt.Sprintf( + "cooldown ends in ~%ds; the next call after that will be a single probe", + int(remain.Seconds()))) + case breakerHalfOpen: + if b.probing { + return true, b.fastFailMessage( + "a probe call is already in flight; await its result before retrying") + } + // Should not happen (half-open implies probing), but allow the call. + b.probing = true + return false, "" + } + return false, "" +} + +// recordFailure registers one infrastructure failure. When the consecutive +// count reaches the threshold, the circuit opens (closed→open) or reopens +// (half-open probe failed → fresh cooldown). +func (b *serverBreaker) recordFailure(err error) { + b.mu.Lock() + defer b.mu.Unlock() + b.failures++ + b.lastErr = errString(err) + switch b.state { + case breakerClosed: + if b.failures >= b.threshold { + b.state = breakerOpen + b.openedAt = time.Now() + b.probing = false + debugLogBreaker(b, "OPEN", "%d consecutive infra failures (last: %s)", b.failures, b.lastErr) + } else { + debugLogBreaker(b, "failure", "%d/%d (last: %s)", b.failures, b.threshold, b.lastErr) + } + case breakerHalfOpen: + // Probe failed: reopen with a fresh cooldown. + b.state = breakerOpen + b.openedAt = time.Now() + b.probing = false + debugLogBreaker(b, "REOPEN", "probe failed: %s", b.lastErr) + case breakerOpen: + // Redundant failure raced the trip; keep the earliest openedAt. + } +} + +// recordSuccess registers one successful (transport-level) call. A server +// semantic error (result.IsError) is still a SUCCESS for breaker purposes: +// the server answered, so it is reachable. +func (b *serverBreaker) recordSuccess() { + b.mu.Lock() + defer b.mu.Unlock() + if b.state != breakerClosed || b.failures > 0 { + debugLogBreaker(b, "CLOSED", "recovered after %d failures", b.failures) + } + b.state = breakerClosed + b.failures = 0 + b.lastErr = "" + b.probing = false +} + +// fastFailMessage renders the actionable guidance given to the model when +// the circuit is open. Mirrors permissionDeniedMessage conventions: +// what happened, why retrying wastes a full LLM round-trip, and what to +// do instead. +func (b *serverBreaker) fastFailMessage(next string) string { + return fmt.Sprintf( + "MCP server %q circuit breaker is OPEN after %d consecutive infrastructure failures "+ + "(last error: %q). All tool calls to this server are being fast-failed WITHOUT "+ + "transport attempts because each retry costs a full agent round-trip that "+ + "cannot succeed while the server is down. %s. Do NOT retry this tool now; "+ + "accomplish the task with other tools, or report the MCP server outage to "+ + "the user and ask them to restart/reconnect the server.", + b.server, b.failures, b.lastErr, next) +} + +// statusSnapshot is a read-only diagnostic view (tests / future TUI status). +type statusSnapshot struct { + State string + Failure int + LastErr string +} + +func (b *serverBreaker) snapshot() statusSnapshot { + b.mu.Lock() + defer b.mu.Unlock() + var s string + switch b.state { + case breakerClosed: + s = "closed" + case breakerOpen: + s = "open" + case breakerHalfOpen: + s = "half-open" + } + return statusSnapshot{State: s, Failure: b.failures, LastErr: b.lastErr} +} + +// debugLogBreaker emits a state-transition log; helper keeps call sites tidy. +func debugLogBreaker(b *serverBreaker, event, format string, args ...interface{}) { + debug.Log("mcp", "breaker[%s]: %s: %s", b.server, event, fmt.Sprintf(format, args...)) +} + +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +// infraErrorPatterns are transport-level failure signatures. Matching is +// lowercase substring against the wrapped error chain rendered by Error(). +// This is intentionally conservative: anything NOT matched is treated as a +// semantic failure and never trips the breaker. +var infraErrorPatterns = []string{ + "connection refused", + "connection reset", + "connection closed", + "broken pipe", + "unexpected eof", + " eof", + "no such host", + "host unreachable", + "network unreachable", + "network is unreachable", + "i/o timeout", + "deadline exceeded", // includes context deadline exceeded (server hang) + "tls:", + "x509", + "http 5", + "status 5", + "server error", + "transport is closing", + "read goroutine did not return after abort", + "pipe is being closed", + "not connected", + "connection timed out", +} + +// isInfraError classifies a CallTool transport error. Semantic errors - +// the server responded with a JSON-RPC error or executed the tool and +// returned isError=true - must NOT trip the breaker. Plain user-initiated +// context cancellation ("context cancelled/canceled", ggcode's Esc path) +// matches NO infra pattern below, so no explicit exclusion is needed; +// transport aborts that EMBED a cancel ("...after abort: context canceled") +// still match infra patterns and are correctly counted as outages. +func isInfraError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + for _, p := range infraErrorPatterns { + if strings.Contains(msg, p) { + return true + } + } + return false +} + +// errNotConnected is recorded when the adapter's caller is nil (the server +// never connected). Treated as infrastructure: no tool on that server can +// work. +type errNotConnected struct{ server string } + +func (e errNotConnected) Error() string { + return fmt.Sprintf("mcp[%s]: caller not connected (server may have crashed or not started)", e.server) +} diff --git a/internal/mcp/breaker_test.go b/internal/mcp/breaker_test.go new file mode 100644 index 000000000..2c9c68784 --- /dev/null +++ b/internal/mcp/breaker_test.go @@ -0,0 +1,261 @@ +package mcp + +// sa-21: per-server MCP circuit breaker tests. + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestIsInfraErrorClassification(t *testing.T) { + infra := []string{ + "mcp[srv]: tools/call x: dial tcp 127.0.0.1:9999: connect: connection refused", + "mcp[srv]: connection closed", + "mcp[srv]: tools/call x: EOF", + "mcp[srv]: tools/call x: read tcp ...: i/o timeout", + "mcp[srv]: tools/call x: context deadline exceeded", + "mcp[srv]: read goroutine did not return after abort: context canceled", + "mcp[srv]: caller not connected (server may have crashed or not started)", + } + for _, m := range infra { + if !isInfraError(errors.New(m)) { + t.Errorf("expected INFRA: %q", m) + } + } + semantic := []string{ + "mcp[srv]: tools/call query: Tool not found: query", + "mcp[srv]: tools/call x: invalid arguments: missing param p", + "mcp[srv]: context cancelled: context canceled", // user pressed Esc + "mcp[srv]: tools/call x: permission denied for path /etc", + } + for _, m := range semantic { + if isInfraError(errors.New(m)) { + t.Errorf("expected SEMANTIC: %q", m) + } + } + if isInfraError(nil) { + t.Error("nil error must not be infra") + } +} + +func TestBreakerOpensAfterThresholdAndFastFails(t *testing.T) { + b := newServerBreaker("srv") + for i := 0; i < b.threshold; i++ { + blocked, _ := b.gate() + if blocked { + t.Fatalf("iteration %d: circuit must be closed", i) + } + b.recordFailure(errors.New("connection refused")) + } + blocked, msg := b.gate() + if !blocked { + t.Fatal("circuit must be OPEN after threshold failures") + } + if !strings.Contains(msg, `"srv"`) || !strings.Contains(msg, "Do NOT retry") { + t.Errorf("fast-fail message missing actionable guidance: %q", msg) + } +} + +func TestBreakerHalfOpenProbeThenClose(t *testing.T) { + b := newServerBreaker("srv") + b.cooldown = 10 * time.Millisecond + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("i/o timeout")) + } + if blocked, _ := b.gate(); !blocked { + t.Fatal("expected OPEN") + } + time.Sleep(15 * time.Millisecond) // cooldown elapses + if blocked, _ := b.gate(); blocked { + t.Fatal("after cooldown the call must become the half-open probe") + } + // Concurrent callers while the probe is in flight are fast-failed. + if blocked, msg := b.gate(); !blocked || !strings.Contains(msg, "probe") { + t.Errorf("expected probe-in-flight fast-fail, got blocked=%v msg=%q", blocked, msg) + } + b.recordSuccess() + if b.snapshot().State != "closed" { + t.Fatalf("success after probe must close, got %s", b.snapshot().State) + } + if b.snapshot().Failure != 0 { + t.Errorf("failure count must reset, got %d", b.snapshot().Failure) + } +} + +func TestBreakerProbeFailureReopens(t *testing.T) { + b := newServerBreaker("srv") + b.cooldown = 5 * time.Millisecond + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("connection reset")) + } + time.Sleep(10 * time.Millisecond) + if blocked, _ := b.gate(); blocked { + t.Fatal("probe must be allowed after cooldown") + } + b.recordFailure(errors.New("connection reset")) + blocked, _ := b.gate() + if !blocked { + t.Fatal("failed probe must reopen the circuit") + } + // Fresh cooldown: the reopened circuit only unlocks after a FRESH + // cooldown, not the original one (10ms elapsed here < 50ms fresh). + b.cooldown = 50 * time.Millisecond + time.Sleep(10 * time.Millisecond) + blocked, msg := b.gate() + if !blocked || !strings.Contains(msg, "OPEN") { + t.Fatalf("reopened circuit must block until a fresh cooldown; got blocked=%v msg=%q", blocked, msg) + } +} + +func TestBreakerSuccessResetsCounter(t *testing.T) { + b := newServerBreaker("srv") + b.recordFailure(errors.New("connection refused")) + b.recordFailure(errors.New("connection refused")) + b.recordSuccess() + b.recordFailure(errors.New("connection refused")) + b.recordFailure(errors.New("connection refused")) + if blocked, _ := b.gate(); blocked { + t.Fatal("success must reset the consecutive-failure counter") + } +} + +func TestMCPTripWireAcrossSiblingTools(t *testing.T) { + // Two tools of the SAME server share one breaker: trips on tool A must + // fast-fail tool B without any transport attempt. + var calls atomic.Int32 + failing := &countingCaller{calls: &calls, err: errors.New("connection refused")} + ok := &countingCaller{calls: &calls} + br := newServerBreaker("srv") + + ta := &mcpTool{name: "mcp__srv__a", caller: failing, toolName: "a", srvName: "srv", breaker: br} + tb := &mcpTool{name: "mcp__srv__b", caller: ok, toolName: "b", srvName: "srv", breaker: br} + + for i := 0; i < br.threshold; i++ { + res, _ := ta.Execute(context.Background(), json.RawMessage(`{}`)) + if !res.IsError { + t.Fatalf("call %d must fail", i) + } + } + if got := calls.Load(); got != int32(br.threshold) { + t.Fatalf("expected exactly %d transport calls, got %d", br.threshold, got) + } + res, _ := tb.Execute(context.Background(), json.RawMessage(`{}`)) + if !res.IsError { + t.Fatal("sibling tool must be fast-failed by the open breaker") + } + if calls.Load() != int32(br.threshold) { + t.Fatalf("sibling fast-fail must NOT attempt transport; calls=%d", calls.Load()) + } + if !strings.Contains(res.Content, "circuit breaker is OPEN") { + t.Errorf("sibling fast-fail must carry breaker message, got: %q", res.Content) + } +} + +func TestMCPSemanticErrorsNeverTripBreaker(t *testing.T) { + var calls atomic.Int32 + failing := &countingCaller{calls: &calls, err: errors.New("mcp[srv]: tools/call x: Tool not found: x")} + br := newServerBreaker("srv") + mt := &mcpTool{name: "mcp__srv__x", caller: failing, toolName: "x", srvName: "srv", breaker: br} + for i := 0; i < br.threshold*2; i++ { + if res, _ := mt.Execute(context.Background(), json.RawMessage(`{}`)); !res.IsError { + t.Fatal("expected error result") + } + } + if got := calls.Load(); got != int32(br.threshold*2) { + t.Fatalf("semantic errors must always reach transport; calls=%d", got) + } + if br.snapshot().State != "closed" { + t.Fatalf("semantic failures must not open the breaker, got %s", br.snapshot().State) + } +} + +func TestAdapterSharesBreakerAcrossRegisteredTools(t *testing.T) { + adapter := NewAdapter("srv", nil, []ToolDefinition{ + {Name: "a"}, {Name: "b"}, + }) + if adapter.breaker == nil { + t.Fatal("NewAdapter must create the per-server breaker") + } + adapter.breaker.threshold = 1 + adapter.breaker.recordFailure(errNotConnected{server: "srv"}) + blocked, msg := adapter.breaker.gate() + if !blocked || !strings.Contains(msg, "OPEN") { + t.Fatalf("expected open circuit, blocked=%v msg=%q", blocked, msg) + } +} + +func TestBreakerConcurrentGateDuringOpen(t *testing.T) { + b := newServerBreaker("srv") + for i := 0; i < b.threshold; i++ { + b.recordFailure(errors.New("connection refused")) + } + var wg sync.WaitGroup + var fastFailed atomic.Int32 + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if blocked, _ := b.gate(); blocked { + fastFailed.Add(1) + } + }() + } + wg.Wait() + if fastFailed.Load() != 50 { + t.Fatalf("all 50 concurrent gates must fast-fail while OPEN, got %d", fastFailed.Load()) + } + if b.snapshot().State != "open" { + t.Fatalf("state must remain open, got %s", b.snapshot().State) + } +} + +func TestBreakerCloneSharesBreaker(t *testing.T) { + br := newServerBreaker("srv") + mt := &mcpTool{name: "mcp__srv__a", caller: nil, toolName: "a", srvName: "srv", breaker: br} + c, ok := mt.Clone().(*mcpTool) + if !ok { + t.Fatal("Clone must return *mcpTool") + } + if c.breaker != br { + t.Fatal("Clone must share the per-server breaker pointer") + } + // nil-caller is infra and feeds the shared breaker. + for i := 0; i < br.threshold; i++ { + c.Execute(context.Background(), json.RawMessage(`{}`)) + } + if br.snapshot().State != "open" { + t.Fatalf("nil-caller failures must trip the breaker, got %s", br.snapshot().State) + } + res, _ := mt.Execute(context.Background(), json.RawMessage(`{}`)) + if !strings.Contains(res.Content, "circuit breaker is OPEN") { + t.Errorf("original tool must also fast-fail, got: %q", res.Content) + } +} + +// countingCaller is a toolCaller that counts transport attempts. +type countingCaller struct { + calls *atomic.Int32 + err error + result *CallToolResult + onCall func() +} + +func (c *countingCaller) CallTool(ctx context.Context, name string, args map[string]interface{}) (*CallToolResult, error) { + c.calls.Add(1) + if c.onCall != nil { + c.onCall() + } + if c.err != nil { + return nil, c.err + } + if c.result != nil { + return c.result, nil + } + return &CallToolResult{}, nil +}