diff --git a/pkg/tools/a2a/a2a.go b/pkg/tools/a2a/a2a.go index c3b812390..6332a4430 100644 --- a/pkg/tools/a2a/a2a.go +++ b/pkg/tools/a2a/a2a.go @@ -14,9 +14,10 @@ import ( "sync" "time" - "github.com/a2aproject/a2a-go/a2a" - "github.com/a2aproject/a2a-go/a2aclient" - "github.com/a2aproject/a2a-go/a2aclient/agentcard" + "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/a2aproject/a2a-go/v2/a2aclient" + "github.com/a2aproject/a2a-go/v2/a2aclient/agentcard" + "github.com/a2aproject/a2a-go/v2/a2acompat/a2av0" "github.com/docker/docker-agent/pkg/config" "github.com/docker/docker-agent/pkg/config/latest" @@ -27,7 +28,7 @@ import ( ) type streamingSender interface { - SendStreamingMessage(ctx context.Context, params *a2a.MessageSendParams) iter.Seq2[a2a.Event, error] + SendStreamingMessage(ctx context.Context, req *a2a.SendMessageRequest) iter.Seq2[a2a.Event, error] } type headerExpander interface { @@ -207,7 +208,8 @@ func (t *Toolset) Start(ctx context.Context) error { rec := &retryAfterRecorder{base: base} client.Transport = rec - resolver := agentcard.NewResolver(client) + // The v0 parser also reads 1.0 cards, so both serve a2a and older A2A servers resolve. + resolver := &agentcard.Resolver{Client: client, CardParser: a2av0.NewAgentCardParser()} card, err := resolver.Resolve(ctx, t.url) if err != nil { return enrichCardError(err, rec) @@ -216,8 +218,8 @@ func (t *Toolset) Start(ctx context.Context) error { httpClient := client endpointOrigin := t.url - if card.URL != "" { - endpointOrigin = card.URL + if len(card.SupportedInterfaces) > 0 && card.SupportedInterfaces[0].URL != "" { + endpointOrigin = card.SupportedInterfaces[0].URL } headers := t.expander.ExpandMap(ctx, t.headers) httpClient.Transport = upstream.NewHeaderTransportForOrigin(base, endpointOrigin, headers) @@ -226,6 +228,7 @@ func (t *Toolset) Start(ctx context.Context) error { ctx, card, a2aclient.WithDefaultsDisabled(), a2aclient.WithJSONRPCTransport(httpClient), + a2av0.WithJSONRPCTransport(a2av0.JSONRPCTransportConfig{Client: httpClient}), ) if err != nil { return fmt.Errorf("failed to create A2A client: %w", err) @@ -266,12 +269,12 @@ func (t *Toolset) createHandler() tools.ToolHandler { return nil, errors.New("A2A client not initialized") } - params := &a2a.MessageSendParams{ - Message: a2a.NewMessage(a2a.MessageRoleUser, &a2a.TextPart{Text: args.Message}), + req := &a2a.SendMessageRequest{ + Message: a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(args.Message)), } var response strings.Builder - for event, err := range client.SendStreamingMessage(ctx, params) { + for event, err := range client.SendStreamingMessage(ctx, req) { if err != nil { return nil, fmt.Errorf("A2A call failed: %w", err) } @@ -308,11 +311,8 @@ func extractText(event a2a.Event) string { var sb strings.Builder for _, part := range parts { - switch p := part.(type) { - case *a2a.TextPart: - sb.WriteString(p.Text) - case a2a.TextPart: - sb.WriteString(p.Text) + if part != nil { + sb.WriteString(part.Text()) } } return sb.String() diff --git a/pkg/tools/a2a/carderror.go b/pkg/tools/a2a/carderror.go index ec339812b..f8bc7f47b 100644 --- a/pkg/tools/a2a/carderror.go +++ b/pkg/tools/a2a/carderror.go @@ -5,7 +5,7 @@ import ( "fmt" "net/http" - "github.com/a2aproject/a2a-go/a2aclient/agentcard" + "github.com/a2aproject/a2a-go/v2/a2aclient/agentcard" "github.com/docker/docker-agent/pkg/modelerrors" ) diff --git a/pkg/tools/a2a/streaming_sender_test.go b/pkg/tools/a2a/streaming_sender_test.go index aa67b4ab1..f913f63bc 100644 --- a/pkg/tools/a2a/streaming_sender_test.go +++ b/pkg/tools/a2a/streaming_sender_test.go @@ -6,17 +6,17 @@ import ( "iter" "testing" - goa2a "github.com/a2aproject/a2a-go/a2a" + goa2a "github.com/a2aproject/a2a-go/v2/a2a" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/docker/docker-agent/pkg/tools" ) -type streamingSenderFunc func(context.Context, *goa2a.MessageSendParams) iter.Seq2[goa2a.Event, error] +type streamingSenderFunc func(context.Context, *goa2a.SendMessageRequest) iter.Seq2[goa2a.Event, error] -func (f streamingSenderFunc) SendStreamingMessage(ctx context.Context, params *goa2a.MessageSendParams) iter.Seq2[goa2a.Event, error] { - return f(ctx, params) +func (f streamingSenderFunc) SendStreamingMessage(ctx context.Context, req *goa2a.SendMessageRequest) iter.Seq2[goa2a.Event, error] { + return f(ctx, req) } func TestStreamingSenderContract(t *testing.T) { @@ -28,22 +28,22 @@ func TestStreamingSenderContract(t *testing.T) { isError bool }{ {name: "empty", want: "No response from agent", isError: true}, - {name: "message", events: []goa2a.Event{goa2a.NewMessage(goa2a.MessageRoleAgent, &goa2a.TextPart{Text: "hello"})}, want: "hello"}, + {name: "message", events: []goa2a.Event{goa2a.NewMessage(goa2a.MessageRoleAgent, goa2a.NewTextPart("hello"))}, want: "hello"}, {name: "mixed", events: []goa2a.Event{ - &goa2a.TaskStatusUpdateEvent{Status: goa2a.TaskStatus{Message: goa2a.NewMessage(goa2a.MessageRoleAgent, &goa2a.TextPart{Text: "one"})}}, - &goa2a.TaskArtifactUpdateEvent{Artifact: &goa2a.Artifact{Parts: goa2a.ContentParts{&goa2a.TextPart{Text: "two"}}}}, - &goa2a.Task{Status: goa2a.TaskStatus{Message: goa2a.NewMessage(goa2a.MessageRoleAgent, &goa2a.TextPart{Text: "three"})}}, + &goa2a.TaskStatusUpdateEvent{Status: goa2a.TaskStatus{Message: goa2a.NewMessage(goa2a.MessageRoleAgent, goa2a.NewTextPart("one"))}}, + &goa2a.TaskArtifactUpdateEvent{Artifact: &goa2a.Artifact{Parts: goa2a.ContentParts{goa2a.NewTextPart("two")}}}, + &goa2a.Task{Status: goa2a.TaskStatus{Message: goa2a.NewMessage(goa2a.MessageRoleAgent, goa2a.NewTextPart("three"))}}, }, want: "onetwothree"}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() calls := 0 tool := NewToolset("test", "https://example.com", nil) - tool.client = streamingSenderFunc(func(ctx context.Context, params *goa2a.MessageSendParams) iter.Seq2[goa2a.Event, error] { + tool.client = streamingSenderFunc(func(ctx context.Context, req *goa2a.SendMessageRequest) iter.Seq2[goa2a.Event, error] { calls++ assert.Equal(t, t.Context(), ctx) - assert.Equal(t, goa2a.MessageRoleUser, params.Message.Role) - assert.Equal(t, "hello", extractText(params.Message)) + assert.Equal(t, goa2a.MessageRoleUser, req.Message.Role) + assert.Equal(t, "hello", extractText(req.Message)) return func(yield func(goa2a.Event, error) bool) { for _, event := range tc.events { if !yield(event, nil) { @@ -66,9 +66,9 @@ func TestStreamingSenderStopsOnError(t *testing.T) { sentinel := errors.New("stream failed") stopped := false tool := NewToolset("test", "https://example.com", nil) - tool.client = streamingSenderFunc(func(context.Context, *goa2a.MessageSendParams) iter.Seq2[goa2a.Event, error] { + tool.client = streamingSenderFunc(func(context.Context, *goa2a.SendMessageRequest) iter.Seq2[goa2a.Event, error] { return func(yield func(goa2a.Event, error) bool) { - if !yield(goa2a.NewMessage(goa2a.MessageRoleAgent, &goa2a.TextPart{Text: "partial"}), nil) { + if !yield(goa2a.NewMessage(goa2a.MessageRoleAgent, goa2a.NewTextPart("partial")), nil) { return } stopped = !yield(nil, sentinel) diff --git a/pkg/tools/a2a/v1_server_test.go b/pkg/tools/a2a/v1_server_test.go new file mode 100644 index 000000000..c80d622da --- /dev/null +++ b/pkg/tools/a2a/v1_server_test.go @@ -0,0 +1,69 @@ +package a2a + +import ( + "context" + "iter" + "net/http" + "net/http/httptest" + "testing" + + a2av1 "github.com/a2aproject/a2a-go/v2/a2a" + a2asrvv1 "github.com/a2aproject/a2a-go/v2/a2asrv" + + "github.com/docker/docker-agent/pkg/tools" +) + +// `docker agent serve a2a` publishes A2A 1.0 cards (supportedInterfaces, no top-level url). +func TestToolSetCallsA2AV1Server(t *testing.T) { + t.Parallel() + + mux := http.NewServeMux() + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + + card := &a2av1.AgentCard{ + Name: "test", + Description: "test", + Version: "1.0.0", + SupportedInterfaces: []*a2av1.AgentInterface{a2av1.NewAgentInterface(server.URL+"/invoke", a2av1.TransportProtocolJSONRPC)}, + Capabilities: a2av1.AgentCapabilities{Streaming: true}, + DefaultInputModes: []string{"text/plain"}, + DefaultOutputModes: []string{"text/plain"}, + Skills: []a2av1.AgentSkill{{ID: "test", Name: "test", Description: "test", Tags: []string{"test"}}}, + } + mux.Handle(a2asrvv1.WellKnownAgentCardPath, a2asrvv1.NewStaticAgentCardHandler(card)) + mux.Handle("/invoke", a2asrvv1.NewJSONRPCHandler(a2asrvv1.NewHandler(testV1Executor{}))) + + toolSet := NewToolset("test", server.URL, nil, WithAllowPrivateIPs(true)) + if err := toolSet.Start(t.Context()); err != nil { + t.Fatalf("Start() error = %v", err) + } + + toolList, err := toolSet.Tools(t.Context()) + if err != nil { + t.Fatalf("Tools() error = %v", err) + } + if len(toolList) != 1 { + t.Fatalf("Tools() returned %d tools, want 1", len(toolList)) + } + + result, err := toolList[0].Handler(t.Context(), tools.ToolCall{Function: tools.FunctionCall{Arguments: `{"message":"hello"}`}}, tools.NopRuntime{}) + if err != nil { + t.Fatalf("Handler() error = %v", err) + } + if result == nil || result.Output != "ok" { + t.Fatalf("Handler() result = %+v, want output %q", result, "ok") + } +} + +type testV1Executor struct{} + +func (testV1Executor) Execute(_ context.Context, execCtx *a2asrvv1.ExecutorContext) iter.Seq2[a2av1.Event, error] { + return func(yield func(a2av1.Event, error) bool) { + yield(a2av1.NewMessageForTask(a2av1.MessageRoleAgent, execCtx, a2av1.NewTextPart("ok")), nil) + } +} + +func (testV1Executor) Cancel(context.Context, *a2asrvv1.ExecutorContext) iter.Seq2[a2av1.Event, error] { + return func(func(a2av1.Event, error) bool) {} +}