diff --git a/go/googleadk/README.md b/go/googleadk/README.md index 107e877d..eab481e6 100644 --- a/go/googleadk/README.md +++ b/go/googleadk/README.md @@ -199,7 +199,24 @@ the model SDK's own retries (see below), or override the fallbacks. toolset name — the toolset is cached and shared across calls — and the plugin closes any cached toolset that implements `Close() error` at worker stop automatically; with manual wiring, call `Activities.Close` yourself after - `worker.Run` returns. + `worker.Run` returns. A tool's own error, such as an MCP `isError` result, + reaches the model once, as it does without Temporal; `CallMcpTool` succeeds + and is not retried. The plugin counts an error as the tool's own when it + wraps no other error and is not an `ApplicationError`. ADK's MCP tools wrap + the error of a failed MCP request, so that error fails `CallMcpTool` with a + retryable `googleadk.McpError`, as does an error returned after the Worker + stops the call. To have a failure in your own toolset or a `FakeMCPServer` + handler retried, wrap its cause with `%w` or return a retryable + `temporal.ApplicationError`. + + If an agent or a runner plugin has a tool-error callback, or an after-tool + callback that treats errors differently from results, don't run Workers of an + earlier version and this one on the same Task Queue. An earlier Workflow + Worker hands this version's tool error to ADK as a result and skips the + callback, so the run fails to replay on the other version or closes as + `Failed`. Deploy this version on a new Task Queue, or stop every earlier + Worker before you start this version's. If you roll back, keep runs that ran + on this version on its Workers until they close. - **Remote A2A agents.** Build ADK's remote agent (`remoteagent.NewA2A`) with `AgentCardProvider: googleadk.NewA2AAgentCardProvider(cardURL, activityOptions)` and `ClientProvider: googleadk.NewA2AClientProvider(activityOptions)`. The card @@ -249,7 +266,11 @@ the model SDK's own retries (see below), or override the fallbacks. serial fallback. - **Typed failures.** Model / tool / MCP / A2A failures surface as Temporal `ApplicationError`s tagged `googleadk.ModelError` / `.ToolError` / `.McpError` - / `.A2AError`. + / `.A2AError`. An MCP tool's own error is not one of these: ADK, its + tool-error callbacks and the model get an error with the tool's message, as + without Temporal, not a `googleadk.McpError`. Its Go type and identity don't + cross the Activity boundary, so `errors.Is` and `errors.As` don't match it; + a callback can only check its message (see **MCP, statelessly** above). Classify with `IsNonRetryable(err)`; never string-match. Upstream HTTP status drives retryability (`408`/`409`/`429`/`5xx` retryable, other `4xx` not). - **Disable model-SDK retries in your `ModelFactory`.** `InvokeModel` already runs diff --git a/go/googleadk/activities.go b/go/googleadk/activities.go index 69b3f09c..e5bc7844 100644 --- a/go/googleadk/activities.go +++ b/go/googleadk/activities.go @@ -62,7 +62,11 @@ type ModelFactory func(ctx context.Context, modelName string) (model.LLM, error) // inside a factory. ctx is only valid for the duration of the constructing // call; mcptoolset.New ignores it — its MCP session opens lazily with each // request's own context. If the returned toolset implements `Close() error`, -// Activities.Close closes it at worker shutdown. +// Activities.Close closes it at worker shutdown. An error a toolset's tool +// returns that wraps no other error and is not an ApplicationError is the +// tool's own answer: CallMcpTool hands it to the model and does not retry it. +// To have a failure retried, wrap its cause with %w or return a retryable +// temporal.ApplicationError. type MCPFactory func(ctx context.Context) (tool.Toolset, error) // Config declares the worker-side registries the Activities consult when a diff --git a/go/googleadk/go.mod b/go/googleadk/go.mod index f58926ef..e1c19082 100644 --- a/go/googleadk/go.mod +++ b/go/googleadk/go.mod @@ -5,6 +5,7 @@ go 1.26.6 require ( github.com/a2aproject/a2a-go/v2 v2.5.0 github.com/google/uuid v1.6.0 + github.com/modelcontextprotocol/go-sdk v1.7.0 github.com/nexus-rpc/sdk-go v0.7.0 github.com/stretchr/testify v1.11.1 go.opentelemetry.io/otel v1.45.0 @@ -23,6 +24,7 @@ require ( require ( cloud.google.com/go v0.123.0 // indirect cloud.google.com/go/auth v0.23.0 // indirect + cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect @@ -44,13 +46,17 @@ require ( github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/robfig/cron v1.2.0 // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.4 // indirect github.com/stretchr/objx v0.5.2 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect go.temporal.io/api v1.63.4 // indirect golang.org/x/crypto v0.55.0 // indirect golang.org/x/mod v0.38.0 // indirect golang.org/x/net v0.58.0 // indirect + golang.org/x/oauth2 v0.36.0 // indirect golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.41.0 // indirect diff --git a/go/googleadk/go.sum b/go/googleadk/go.sum index 4a90a131..cfe79e87 100644 --- a/go/googleadk/go.sum +++ b/go/googleadk/go.sum @@ -2,6 +2,8 @@ cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= cloud.google.com/go/auth v0.23.0 h1:6Gg1CMgpgubRG7DGz5Vf1pcoNo8RfiRiRAPS4crTp54= cloud.google.com/go/auth v0.23.0/go.mod h1:4DhBRcqvtljQN3dJ57qtqbib5ZGCYE5f2crfiiC2EM0= +cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= +cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= github.com/a2aproject/a2a-go/v2 v2.5.0 h1:ZdcFoxv+nZTUV0i2ue5hES76YCANFPG9vjqd7vK8yWM= @@ -21,6 +23,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc= github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= @@ -51,6 +55,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/modelcontextprotocol/go-sdk v1.7.0 h1:yqjY2dsbKAC0LSuWZVBMrHgiG8ukXv6NRo0JiALay44= +github.com/modelcontextprotocol/go-sdk v1.7.0/go.mod h1:dL7u98E/zjJTGzEq+j30jQ8K2k1mb6LeAH4inEcSGts= github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80= github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y= github.com/nexus-rpc/sdk-go v0.7.0 h1:38NrfY5rLnZAiMMs2ZfCKI/CSDzdfJG+27iAgfA8bUI= @@ -61,15 +67,23 @@ github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ= github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 h1:0Qx7VGBacMm9ZENQ7TnNObTYI4ShC+lHI16seduaxZo= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0/go.mod h1:Sje3i3MjSPKTSPvVWCaL8ugBzJwik3u4smCjUeuupqg= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 h1:CqXxU8VOmDefoh0+ztfGaymYbhdB/tT3zs79QaZTNGY= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0/go.mod h1:BuhAPThV8PBHBvg8ZzZ/Ok3idOdhWIodywz2xEcRbJo= go.opentelemetry.io/otel v1.45.0 h1:pdrWmLHofpubmArBv1LgFSv1Z0Ie/ppdZzu+kUN5EeU= @@ -113,6 +127,8 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= +golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -139,6 +155,8 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= +golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/go/googleadk/mcp.go b/go/googleadk/mcp.go index c6ce2a2d..bc739858 100644 --- a/go/googleadk/mcp.go +++ b/go/googleadk/mcp.go @@ -146,6 +146,13 @@ func (t *mcpProxyTool) Run(ctx agent.Context, args any) (map[string]any, error) } return nil, fmt.Errorf("error tool %q %w", t.Name(), tool.ErrConfirmationRequired) } + if out.Error != nil { + // The tool's own error, such as an MCP isError result. Give ADK an + // error with the tool's message, so callbacks and the model see it as + // they do without Temporal. Only the message crosses the Activity + // boundary, not the error's Go type. + return nil, errors.New(*out.Error) + } if out.Result == nil { return map[string]any{}, nil } @@ -168,10 +175,14 @@ type mcpCallInput struct { // mcpCallOutput is the serializable result of CallMcpTool. Confirmation is // non-nil when the tool paused on ADK's tool.ErrConfirmationRequired protocol // instead of running; the workflow-side proxy re-records it so the runner -// surfaces the pause exactly like an in-workflow tool. +// surfaces the pause exactly like an in-workflow tool. Error is the message of +// the tool's own error, which the proxy hands to ADK; see CallMcpTool. Result +// then holds {"error": message}, the response ADK builds from a tool error, so +// a workflow worker without the Error field gives the model the same response. type mcpCallOutput struct { Result map[string]any Confirmation *toolconfirmation.ToolConfirmation + Error *string } // ListMcpTools returns the full declarations (name + description + parameters) @@ -211,6 +222,16 @@ func (a *Activities) ListMcpTools(ctx context.Context, in mcpListInput) ([]*gena // confirmation protocol (e.g. mcptoolset's RequireConfirmation option) the // Activity succeeds with the pending confirmation in its output rather than // failing, so the workflow can pause the agent for the human. +// +// A tool's own error, one that wraps no other error and is not an +// ApplicationError, also goes in the output, so the model sees it once, as it +// does without Temporal. ADK's MCP tools return an isError result (and, before +// ADK v2.4.0, a result with no text) that way. They wrap the error of a failed +// MCP request, which still fails the Activity with a retryable McpError, as +// does an error returned after the Activity's context is done, such as when +// the Worker stops. An ApplicationError returned as is fails the Activity with +// its own type and retryability; wrapped in another error, it fails as a +// retryable wrapError. func (a *Activities) CallMcpTool(ctx context.Context, in mcpCallInput) (*mcpCallOutput, error) { log := activity.GetLogger(ctx) // No-op unless the user set a HeartbeatTimeout on the toolset's Activity @@ -256,6 +277,12 @@ func (a *Activities) CallMcpTool(ctx context.Context, in mcpCallInput) (*mcpCall return nil, newApplicationError(ErrorTypeMCP, false, runErr, "tool %q failed: %v", in.Tool, runErr) } + // An error after ctx is done, such as ctx.Err() when the Worker + // stops, is not the tool's answer, whatever it wraps. + if ctx.Err() == nil && isToolAnswer(runErr) { + msg := runErr.Error() + return &mcpCallOutput{Result: map[string]any{"error": msg}, Error: &msg}, nil + } return nil, classifyToolError(ErrorTypeMCP, in.Tool, runErr) } return &mcpCallOutput{Result: res}, nil @@ -264,6 +291,23 @@ func (a *Activities) CallMcpTool(ctx context.Context, in mcpCallInput) (*mcpCall "tool %q not found in MCP toolset %q", in.Tool, in.Toolset) } +// isToolAnswer reports whether err, returned by an MCP tool's Run, is the +// tool's own answer: an error that wraps no other error and is not an +// ApplicationError. +func isToolAnswer(err error) bool { + var appErr interface{ NonRetryable() bool } + if errors.As(err, &appErr) { + return false + } + switch e := err.(type) { + case interface{ Unwrap() error }: + return e.Unwrap() == nil + case interface{ Unwrap() []error }: + return len(e.Unwrap()) == 0 + } + return true +} + // mcpToolset resolves the named toolset, running its registered factory at most // once and caching the result for the life of the Activities value. The cache // is what bounds the resource cost: a live MCP toolset lazily opens and retains diff --git a/go/googleadk/mcp_error_test.go b/go/googleadk/mcp_error_test.go new file mode 100644 index 00000000..82eb0cad --- /dev/null +++ b/go/googleadk/mcp_error_test.go @@ -0,0 +1,217 @@ +package googleadk_test + +import ( + "context" + "errors" + "regexp" + "sync/atomic" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.temporal.io/sdk/testsuite" + "go.temporal.io/sdk/worker" + "go.temporal.io/sdk/workflow" + + "google.golang.org/genai" + + "google.golang.org/adk/v2/agent" + "google.golang.org/adk/v2/agent/llmagent" + "google.golang.org/adk/v2/runner" + "google.golang.org/adk/v2/session" + "google.golang.org/adk/v2/tool" + "google.golang.org/adk/v2/tool/mcptoolset" + + "go.temporal.io/googleadk" +) + +// mcpLookupWorkflow runs an agent with the "orders" MCP toolset, two attempts +// per MCP Activity, and returns the "lookup" tool response the model saw. Its +// tool-error callback hands the model the error it got, with "callback": true. +func mcpLookupWorkflow(ctx workflow.Context) (map[string]any, error) { + root, err := llmagent.New(llmagent.Config{ + Name: "assistant", + Model: googleadk.NewModel("fake-model"), + Instruction: "look up orders", + OnToolErrorCallbacks: []llmagent.OnToolErrorCallback{ + func(_ agent.Context, _ tool.Tool, _ map[string]any, err error) (map[string]any, error) { + return map[string]any{"error": err.Error(), "callback": true}, nil + }, + }, + Toolsets: []tool.Toolset{googleadk.NewMCPToolset(googleadk.MCPToolsetOptions{ + Name: "orders", + ActivityOptions: workflow.ActivityOptions{ + RetryPolicy: &temporal.RetryPolicy{InitialInterval: time.Millisecond, MaximumAttempts: 2}, + }, + })}, + }) + if err != nil { + return nil, err + } + r, err := runner.New(runner.Config{AppName: "test-app", Agent: root, SessionService: session.InMemoryService(), AutoCreateSession: true}) + if err != nil { + return nil, err + } + var res runResult + msg := genai.NewContentFromText("look up order A-1", genai.RoleUser) + for ev, err := range r.Run(googleadk.NewContext(ctx), "user-1", "session-1", msg, agent.RunConfig{}) { + if err != nil { + return nil, err + } + collectEvent(&res, ev) + } + return res.ToolResponsePayloads["lookup"], nil +} + +type lookupArgs struct { + Order string `json:"order"` +} + +// adkLookup returns a factory for ADK's mcptoolset over an in-memory go-sdk MCP +// server whose "lookup" tool always returns an isError result. With +// failFirst, the server answers the first tools/call request with a JSON-RPC +// error, like a failed request. +func adkLookup(failFirst bool) func(runs *atomic.Int32) googleadk.MCPFactory { + return func(runs *atomic.Int32) googleadk.MCPFactory { + return func(context.Context) (tool.Toolset, error) { + srv := mcp.NewServer(&mcp.Implementation{Name: "orders", Version: "1"}, nil) + mcp.AddTool(srv, &mcp.Tool{Name: "lookup", Description: "Look up an order."}, + func(_ context.Context, _ *mcp.CallToolRequest, in lookupArgs) (*mcp.CallToolResult, any, error) { + runs.Add(1) + return &mcp.CallToolResult{IsError: true, Content: []mcp.Content{&mcp.TextContent{Text: "order " + in.Order + " not found"}}}, nil, nil + }) + var failed atomic.Bool + srv.AddReceivingMiddleware(func(next mcp.MethodHandler) mcp.MethodHandler { + return func(ctx context.Context, method string, req mcp.Request) (mcp.Result, error) { + if failFirst && method == "tools/call" && failed.CompareAndSwap(false, true) { + return nil, errors.New("server unavailable") + } + return next(ctx, method, req) + } + }) + clientTransport, serverTransport := mcp.NewInMemoryTransports() + if _, err := srv.Connect(context.Background(), serverTransport, nil); err != nil { + return nil, err + } + return mcptoolset.New(mcptoolset.Config{Transport: clientTransport}) + } + } +} + +// fakeLookup returns a factory for a FakeMCPServer whose "lookup" handler +// always returns err. +func fakeLookup(err error) func(runs *atomic.Int32) googleadk.MCPFactory { + return func(runs *atomic.Int32) googleadk.MCPFactory { + return googleadk.NewFakeMCPServer("orders").AddTool("lookup", "Look up an order.", nil, + func(map[string]any) (map[string]any, error) { + runs.Add(1) + return nil, err + }).Factory() + } +} + +// TestMcpToolErrorReachesModel proves an MCP tool's own error, such as an +// isError result, reaches the tool-error callback and the model once with the +// text ADK gives it without Temporal, while a failed MCP request, or an error +// that wraps several errors, is still retried and an ApplicationError still +// fails the Activity. It also decodes CallMcpTool's +// output as a workflow worker from before the fix does, reading only Result, +// and checks that it holds the same error. +func TestMcpToolErrorReachesModel(t *testing.T) { + notFound := "^" + regexp.QuoteMeta("Tool execution failed. Details: order A-1 not found") + "$" + for _, tc := range []struct { + name string + factory func(runs *atomic.Int32) googleadk.MCPFactory + wantCalls int // CallMcpTool attempts + wantRuns int32 // tool runs + want string // pattern for the "error" the callback gets and the model sees + answered bool // CallMcpTool succeeds with the tool's error + }{ + {"isError result", adkLookup(false), 1, 1, notFound, true}, + {"failed request, then isError result", adkLookup(true), 2, 1, notFound, true}, + {"FakeMCPServer handler error", fakeLookup(errors.New("order A-1 not found")), 1, 1, "^order A-1 not found$", true}, + {"FakeMCPServer handler joined errors", fakeLookup(errors.Join(errors.New("order A-1"), errors.New("not found"))), 2, 2, + `(?s)^activity error \(type: googleadk\.CallMcpTool, .*\(type: googleadk\.McpError, retryable: true\)`, false}, + {"ApplicationError", fakeLookup(temporal.NewNonRetryableApplicationError("order A-1 not found", "NotFound", nil)), 1, 1, + `^activity error \(type: googleadk\.CallMcpTool, .*\(type: NotFound, retryable: false\)$`, false}, + } { + t.Run(tc.name, func(t *testing.T) { + var runs atomic.Int32 + var s testsuite.WorkflowTestSuite + env := s.NewTestWorkflowEnvironment() + env.RegisterWorkflow(mcpLookupWorkflow) + counter := wireActivities(t, env, googleadk.Config{ + Models: map[string]googleadk.ModelFactory{"fake-model": scriptedModelFactory( + googleadk.FunctionCallResponse("c1", "lookup", map[string]any{"order": "A-1"}), + googleadk.TextResponse("done"), + )}, + MCPToolsets: map[string]googleadk.MCPFactory{"orders": tc.factory(&runs)}, + }) + // The fields a workflow worker from before the fix decodes. + var oldOut *struct{ Result map[string]any } + env.SetOnActivityCompletedListener(func(info *activity.Info, result converter.EncodedValue, err error) { + if info.ActivityType.Name == googleadk.CallMcpToolActivityName && err == nil { + oldOut = new(struct{ Result map[string]any }) + require.NoError(t, result.Get(oldOut)) + } + }) + + env.ExecuteWorkflow(mcpLookupWorkflow) + + require.True(t, env.IsWorkflowCompleted()) + require.NoError(t, env.GetWorkflowError()) + var got map[string]any + require.NoError(t, env.GetWorkflowResult(&got)) + assert.Regexp(t, tc.want, got["error"]) + assert.Equal(t, true, got["callback"], "tool-error callback") + assert.Equal(t, tc.wantRuns, runs.Load(), "tool runs") + assert.Equal(t, tc.wantCalls, counter.get(googleadk.CallMcpToolActivityName), "CallMcpTool attempts") + if tc.answered { + require.NotNil(t, oldOut, "CallMcpTool output") + assert.Equal(t, map[string]any{"error": got["error"]}, oldOut.Result, "Result read by an older workflow worker") + } else { + assert.Nil(t, oldOut, "CallMcpTool output") + } + }) + } +} + +// TestMcpToolErrorAfterWorkerStops proves an error a tool returns after the +// Activity's context is cancelled, as when the Worker stops during the call, +// fails CallMcpTool with a retryable McpError, so the call runs again, instead +// of reaching the model as the tool's answer. The workflow test environment +// cannot show it: it has no Worker stop. +func TestMcpToolErrorAfterWorkerStops(t *testing.T) { + bg, stop := context.WithCancel(context.Background()) + defer stop() + lookup, err := funcTool("lookup", func(ctx agent.Context, _ map[string]any) (map[string]any, error) { + stop() // the Worker stops during the call + <-ctx.Done() + return nil, ctx.Err() + }) + require.NoError(t, err) + acts, err := googleadk.NewActivities(googleadk.Config{MCPToolsets: map[string]googleadk.MCPFactory{ + "orders": func(context.Context) (tool.Toolset, error) { + return staticToolset{name: "orders", tools: []tool.Tool{lookup}}, nil + }, + }}) + require.NoError(t, err) + var s testsuite.WorkflowTestSuite + env := s.NewTestActivityEnvironment() + env.SetWorkerOptions(worker.Options{BackgroundActivityContext: bg}) + env.RegisterActivityWithOptions(acts.CallMcpTool, activity.RegisterOptions{Name: googleadk.CallMcpToolActivityName}) + + _, err = env.ExecuteActivity(googleadk.CallMcpToolActivityName, + map[string]any{"Toolset": "orders", "Tool": "lookup", "Args": map[string]any{}}) + + var appErr *temporal.ApplicationError + require.ErrorAs(t, err, &appErr, "CallMcpTool must fail, not return the tool's answer") + assert.Equal(t, googleadk.ErrorTypeMCP, appErr.Type()) + assert.False(t, appErr.NonRetryable()) +} diff --git a/go/googleadk/testing.go b/go/googleadk/testing.go index 925e1879..922d3b08 100644 --- a/go/googleadk/testing.go +++ b/go/googleadk/testing.go @@ -98,7 +98,11 @@ func NewFakeMCPServer(name string) *FakeMCPServer { } // AddTool registers a tool on the fake server. handler is invoked when the tool -// runs worker-side; schema describes its parameters (may be nil). +// runs worker-side; schema describes its parameters (may be nil). An error +// handler returns that wraps no other error and is not an ApplicationError is +// the tool's own answer: it reaches the model and is not retried. To have a +// failure retried, wrap its cause with %w or return a retryable +// temporal.ApplicationError. func (s *FakeMCPServer) AddTool(name, description string, schema *genai.Schema, handler func(args map[string]any) (map[string]any, error)) *FakeMCPServer { s.tools = append(s.tools, &fakeMCPTool{ decl: &genai.FunctionDeclaration{Name: name, Description: description, Parameters: schema},