From 62e5f457d23f529256739087d6620d87ae82b4fd Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 09:15:48 +0000 Subject: [PATCH 01/16] api: add the a2a service type --- api/network.go | 7 +++++++ api/network_test.go | 17 +++++++++++++++++ api/sam.pb.go | 8 ++++++-- api/sam.proto | 1 + 4 files changed, 31 insertions(+), 2 deletions(-) diff --git a/api/network.go b/api/network.go index 53afa43a..ec6a86bf 100644 --- a/api/network.go +++ b/api/network.go @@ -171,6 +171,9 @@ const ( // ServiceTypeStringInference is the string identifier for Inference services. ServiceTypeStringInference = "inference" + + // ServiceTypeStringA2A is the string identifier for A2A (Agent2Agent) services. + ServiceTypeStringA2A = "a2a" ) // ParseServiceType converts a string identifier (e.g. from JSON or REST) to the ServiceType protobuf enum. @@ -180,6 +183,8 @@ func ParseServiceType(s string) (ServiceType, error) { return ServiceType_SERVICE_TYPE_MCP, nil case ServiceTypeStringInference: return ServiceType_SERVICE_TYPE_INFERENCE, nil + case ServiceTypeStringA2A: + return ServiceType_SERVICE_TYPE_A2A, nil default: return ServiceType_SERVICE_TYPE_UNSPECIFIED, fmt.Errorf("invalid service type: %s", s) } @@ -192,6 +197,8 @@ func ServiceTypeToString(t ServiceType) (string, error) { return ServiceTypeStringMCP, nil case ServiceType_SERVICE_TYPE_INFERENCE: return ServiceTypeStringInference, nil + case ServiceType_SERVICE_TYPE_A2A: + return ServiceTypeStringA2A, nil default: return "", fmt.Errorf("invalid or unspecified service type") } diff --git a/api/network_test.go b/api/network_test.go index 6e40187f..b33b40f9 100644 --- a/api/network_test.go +++ b/api/network_test.go @@ -99,3 +99,20 @@ func TestParseServiceTarget(t *testing.T) { }) } } + +func TestServiceTypeA2ARoundTrip(t *testing.T) { + st, err := ParseServiceType("a2a") + if err != nil { + t.Fatalf("ParseServiceType(a2a): %v", err) + } + if st != ServiceType_SERVICE_TYPE_A2A { + t.Fatalf("ParseServiceType(a2a) = %v, want SERVICE_TYPE_A2A", st) + } + s, err := ServiceTypeToString(st) + if err != nil { + t.Fatalf("ServiceTypeToString: %v", err) + } + if s != ServiceTypeStringA2A { + t.Fatalf("ServiceTypeToString = %q, want %q", s, ServiceTypeStringA2A) + } +} diff --git a/api/sam.pb.go b/api/sam.pb.go index deed7a67..8bf34004 100644 --- a/api/sam.pb.go +++ b/api/sam.pb.go @@ -93,6 +93,7 @@ const ( ServiceType_SERVICE_TYPE_UNSPECIFIED ServiceType = 0 ServiceType_SERVICE_TYPE_MCP ServiceType = 1 ServiceType_SERVICE_TYPE_INFERENCE ServiceType = 2 + ServiceType_SERVICE_TYPE_A2A ServiceType = 3 ) // Enum value maps for ServiceType. @@ -101,11 +102,13 @@ var ( 0: "SERVICE_TYPE_UNSPECIFIED", 1: "SERVICE_TYPE_MCP", 2: "SERVICE_TYPE_INFERENCE", + 3: "SERVICE_TYPE_A2A", } ServiceType_value = map[string]int32{ "SERVICE_TYPE_UNSPECIFIED": 0, "SERVICE_TYPE_MCP": 1, "SERVICE_TYPE_INFERENCE": 2, + "SERVICE_TYPE_A2A": 3, } ) @@ -3073,11 +3076,12 @@ const file_api_sam_proto_rawDesc = "" + "\x1dENROLLMENT_STATUS_UNSPECIFIED\x10\x00\x12\x1d\n" + "\x19ENROLLMENT_STATUS_PENDING\x10\x01\x12\x1e\n" + "\x1aENROLLMENT_STATUS_APPROVED\x10\x02\x12\x1e\n" + - "\x1aENROLLMENT_STATUS_REJECTED\x10\x03*]\n" + + "\x1aENROLLMENT_STATUS_REJECTED\x10\x03*s\n" + "\vServiceType\x12\x1c\n" + "\x18SERVICE_TYPE_UNSPECIFIED\x10\x00\x12\x14\n" + "\x10SERVICE_TYPE_MCP\x10\x01\x12\x1a\n" + - "\x16SERVICE_TYPE_INFERENCE\x10\x02B\x1bZ\x19github.com/google/sam/apib\x06proto3" + "\x16SERVICE_TYPE_INFERENCE\x10\x02\x12\x14\n" + + "\x10SERVICE_TYPE_A2A\x10\x03B\x1bZ\x19github.com/google/sam/apib\x06proto3" var ( file_api_sam_proto_rawDescOnce sync.Once diff --git a/api/sam.proto b/api/sam.proto index 458cd895..a368199c 100644 --- a/api/sam.proto +++ b/api/sam.proto @@ -103,6 +103,7 @@ enum ServiceType { SERVICE_TYPE_UNSPECIFIED = 0; SERVICE_TYPE_MCP = 1; SERVICE_TYPE_INFERENCE = 2; + SERVICE_TYPE_A2A = 3; } message ServiceInfo { From 43eeedb08c819eac60806f63a9069c6c863587b3 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 09:21:28 +0000 Subject: [PATCH 02/16] node: accept a2a services with URL backends --- internal/node/a2a_service.go | 43 ++++++++++++++++++++++ internal/node/a2a_service_test.go | 59 +++++++++++++++++++++++++++++++ internal/node/service.go | 7 ++-- 3 files changed, 104 insertions(+), 5 deletions(-) create mode 100644 internal/node/a2a_service.go create mode 100644 internal/node/a2a_service_test.go diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go new file mode 100644 index 00000000..5bf8af53 --- /dev/null +++ b/internal/node/a2a_service.go @@ -0,0 +1,43 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "context" + "fmt" + + "github.com/google/sam/api" +) + +// A2AService proxies Agent2Agent (A2A) JSON-RPC/REST traffic to a local +// agent process. URL backends only: a command backend would wire the A2A +// route to an MCP stdio bridge no A2A client can talk to. +type A2AService struct{ baseService } + +func (s *A2AService) Init(ctx context.Context) error { + switch x := s.backend.(type) { + case *api.RegisterServiceRequest_TargetUrl: + h, err := newReverseProxyHandler(x.TargetUrl) + if err != nil { + return err + } + s.handler = h + case *api.RegisterServiceRequest_Command: + return fmt.Errorf("command-based backends are not supported for A2AService") + default: + return fmt.Errorf("unsupported backend type %T for A2AService", s.backend) + } + return nil +} diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go new file mode 100644 index 00000000..0a0611f6 --- /dev/null +++ b/internal/node/a2a_service_test.go @@ -0,0 +1,59 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "context" + "testing" + + "github.com/google/sam/api" +) + +func TestA2AServiceInitRejectsCommand(t *testing.T) { + svc := &A2AService{baseService: baseService{ + info: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_A2A, Name: "agent"}, + backend: &api.RegisterServiceRequest_Command{Command: &api.CommandBackend{Command: []string{"echo"}}}, + }} + if err := svc.Init(context.Background()); err == nil { + t.Fatal("command backend must be rejected for a2a services") + } +} + +func TestA2AServiceInitURLBackend(t *testing.T) { + svc := &A2AService{baseService: baseService{ + info: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_A2A, Name: "agent"}, + backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: "http://127.0.0.1:9999"}, + }} + if err := svc.Init(context.Background()); err != nil { + t.Fatalf("url backend must be accepted: %v", err) + } + if svc.Handler() == nil { + t.Fatal("nil handler after Init") + } +} + +func TestNewServiceFromRequestA2A(t *testing.T) { + req := &api.RegisterServiceRequest{ + Service: &api.ServiceInfo{Type: api.ServiceType_SERVICE_TYPE_A2A, Name: "agent"}, + Backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: "http://127.0.0.1:9999"}, + } + svc, err := NewServiceFromRequest(req) + if err != nil { + t.Fatalf("factory must accept a2a: %v", err) + } + if _, ok := svc.(*A2AService); !ok { + t.Fatalf("factory returned %T, want *A2AService", svc) + } +} diff --git a/internal/node/service.go b/internal/node/service.go index 47406c83..5ab02214 100644 --- a/internal/node/service.go +++ b/internal/node/service.go @@ -118,11 +118,6 @@ func (b *baseService) Teardown() error { return b.cmd.Process.Kill() } -// A2AService is zero-override embedding. They exist -// so the factory produces a distinct type per ServiceType, leaving room for -// future per-kind behaviour without churn. -type A2AService struct{ baseService } - func NewServiceFromRequest(req *api.RegisterServiceRequest) (Service, error) { info := req.Service switch info.Type { @@ -130,6 +125,8 @@ func NewServiceFromRequest(req *api.RegisterServiceRequest) (Service, error) { return &MCPService{baseService: baseService{info: info, backend: req.Backend}}, nil case api.ServiceType_SERVICE_TYPE_INFERENCE: return &InferenceService{baseService: baseService{info: info, backend: req.Backend}}, nil + case api.ServiceType_SERVICE_TYPE_A2A: + return &A2AService{baseService: baseService{info: info, backend: req.Backend}}, nil default: return nil, fmt.Errorf("unspecified or unsupported service type: %v", info.Type) } From 3e9c515b6178e6ecdf79357fa26ecea6b65678de Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 09:29:24 +0000 Subject: [PATCH 03/16] node: gate raw a2a egress with labels and rewrite agent cards for mesh use --- internal/node/a2a_service.go | 113 +++++++++++++++++++++++++++++ internal/node/a2a_service_test.go | 115 ++++++++++++++++++++++++++++++ internal/node/sidecar.go | 8 ++- 3 files changed, 235 insertions(+), 1 deletion(-) diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index 5bf8af53..f5429b5a 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -15,10 +15,17 @@ package node import ( + "bytes" "context" + "encoding/json" "fmt" + "io" + "net/http" + "strconv" + "strings" "github.com/google/sam/api" + "github.com/libp2p/go-libp2p/core/peer" ) // A2AService proxies Agent2Agent (A2A) JSON-RPC/REST traffic to a local @@ -41,3 +48,109 @@ func (s *A2AService) Init(ctx context.Context) error { } return nil } + +// a2aCardBaseURL is the context key carrying the caller-facing mesh base URL +// of an agent-card fetch, set by a2aEgressHook and consumed by the rewrite. +type a2aCardBaseURL struct{} + +// a2aEgressHook runs the caller-side A2A checks on a raw egress request: +// the fail-closed labels gate and tagging agent-card fetches for rewrite. +// On refusal it writes the HTTP error itself and returns ok=false. +func a2aEgressHook(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { + parts := strings.SplitN(r.URL.Path, "/", 6) + if len(parts) < 5 || parts[3] != api.ServiceTypeStringA2A { + return r, true + } + if labelsHeader := r.Header.Get(api.HeaderSamRequiredLabels); labelsHeader != "" { + r.Header.Del(api.HeaderSamRequiredLabels) + required, err := parseRequiredLabels(labelsHeader) + if err != nil { + http.Error(w, fmt.Sprintf("Invalid %s header: %v", api.HeaderSamRequiredLabels, err), http.StatusBadRequest) + return r, false + } + pid, err := peer.Decode(parts[2]) + if err != nil { + http.Error(w, "Invalid peer ID", http.StatusBadRequest) + return r, false + } + if err := node.VerifyPeerLabels(r.Context(), pid, required); err != nil { + logger.Warnf("[A2A] label gate refused egress to %s: %v", parts[2], err) + http.Error(w, "Required labels not attested by provider", http.StatusForbidden) + return r, false + } + } + if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/.well-known/agent-card.json") { + base := fmt.Sprintf("http://%s/sam/%s/%s/%s", r.Host, parts[2], parts[3], parts[4]) + r = r.WithContext(context.WithValue(r.Context(), a2aCardBaseURL{}, base)) + } + return r, true +} + +// rewriteA2AAgentCard makes a proxied agent card usable by stock A2A clients: +// interface URLs point back at the mesh path, transports the mesh cannot +// carry are dropped, and streaming is advertised off until verified. +func rewriteA2AAgentCard(resp *http.Response) error { + base, ok := resp.Request.Context().Value(a2aCardBaseURL{}).(string) + if !ok || resp.StatusCode != http.StatusOK { + return nil + } + if resp.Header.Get("Content-Encoding") != "" { + logger.Warnf("[A2A] agent card response is content-encoded; skipping rewrite") + return nil + } + body, err := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if err != nil { + return err + } + var card map[string]any + if err := json.Unmarshal(body, &card); err != nil { + return fmt.Errorf("agent card is not valid JSON: %w", err) + } + if _, ok := card["url"]; ok { + card["url"] = base + } + if pt, ok := card["preferredTransport"].(string); ok && !a2aTransportOverHTTP(pt) { + card["preferredTransport"] = "JSONRPC" + } + for _, key := range []string{"additionalInterfaces", "supportedInterfaces"} { + ifaces, ok := card[key].([]any) + if !ok { + continue + } + kept := make([]any, 0, len(ifaces)) + for _, entry := range ifaces { + iface, ok := entry.(map[string]any) + if !ok { + continue + } + transport, _ := iface["transport"].(string) + if transport == "" { + transport, _ = iface["protocolBinding"].(string) + } + if !a2aTransportOverHTTP(transport) { + continue + } + iface["url"] = base + kept = append(kept, iface) + } + card[key] = kept + } + if caps, ok := card["capabilities"].(map[string]any); ok { + caps["streaming"] = false + } + out, err := json.Marshal(card) + if err != nil { + return err + } + resp.Body = io.NopCloser(bytes.NewReader(out)) + resp.ContentLength = int64(len(out)) + resp.Header.Set("Content-Length", strconv.Itoa(len(out))) + return nil +} + +// a2aTransportOverHTTP reports whether an A2A transport can traverse the +// mesh's HTTP-over-libp2p path; gRPC needs its own end-to-end connection. +func a2aTransportOverHTTP(transport string) bool { + return transport == "JSONRPC" || transport == "HTTP+JSON" +} diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index 0a0611f6..2418b3d9 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -16,6 +16,11 @@ package node import ( "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" "testing" "github.com/google/sam/api" @@ -57,3 +62,113 @@ func TestNewServiceFromRequestA2A(t *testing.T) { t.Fatalf("factory returned %T, want *A2AService", svc) } } + +func TestA2AEgressHookNonA2APassthrough(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/mcp/svc/foo", nil) + req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") + _, ok := a2aEgressHook(nil, rec, req) + if !ok { + t.Fatal("non-a2a path must pass through") + } + if req.Header.Get(api.HeaderSamRequiredLabels) == "" { + t.Fatal("labels header on non-a2a path must be left untouched") + } +} + +func TestA2AEgressHookMalformedLabels(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/sam/12D3KooWpeer/a2a/agent/", nil) + req.Header.Set(api.HeaderSamRequiredLabels, "not-a-label") + _, ok := a2aEgressHook(nil, rec, req) + if ok { + t.Fatal("malformed labels must be refused") + } + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", rec.Code) + } +} + +func TestA2AEgressHookTagsCardFetch(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + req.Host = "127.0.0.1:8080" + r2, ok := a2aEgressHook(nil, rec, req) + if !ok { + t.Fatal("card fetch must pass through") + } + base, _ := r2.Context().Value(a2aCardBaseURL{}).(string) + want := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + if base != want { + t.Fatalf("card base = %q, want %q", base, want) + } +} + +func TestRewriteA2AAgentCard(t *testing.T) { + card := `{ + "name": "T", + "url": "http://localhost:9999", + "preferredTransport": "GRPC", + "additionalInterfaces": [ + {"url": "http://localhost:9999", "transport": "JSONRPC"}, + {"url": "localhost:50051", "transport": "GRPC"} + ], + "supportedInterfaces": [ + {"url": "http://localhost:9999", "protocolBinding": "JSONRPC"}, + {"url": "localhost:50051", "protocolBinding": "GRPC"} + ], + "capabilities": {"streaming": true} + }` + base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(card)), + Request: req, + } + if err := rewriteA2AAgentCard(resp); err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + var got map[string]any + if err := json.Unmarshal(body, &got); err != nil { + t.Fatalf("rewritten card is not JSON: %v", err) + } + if got["url"] != base { + t.Errorf("url = %v, want %s", got["url"], base) + } + if got["preferredTransport"] != "JSONRPC" { + t.Errorf("preferredTransport = %v, want JSONRPC", got["preferredTransport"]) + } + for _, key := range []string{"additionalInterfaces", "supportedInterfaces"} { + ifaces, _ := got[key].([]any) + if len(ifaces) != 1 { + t.Fatalf("%s: want 1 HTTP interface after dropping gRPC, got %v", key, got[key]) + } + if u := ifaces[0].(map[string]any)["url"]; u != base { + t.Errorf("%s url = %v, want %s", key, u, base) + } + } + if s := got["capabilities"].(map[string]any)["streaming"]; s != false { + t.Errorf("streaming = %v, want false", s) + } +} + +func TestRewriteA2AAgentCardNoopWithoutTag(t *testing.T) { + orig := `{"name":"T","url":"http://localhost:9999"}` + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(orig)), + Request: httptest.NewRequest("GET", "/anything", nil), + } + if err := rewriteA2AAgentCard(resp); err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + if string(body) != orig { + t.Fatalf("untagged response was modified: %s", body) + } +} diff --git a/internal/node/sidecar.go b/internal/node/sidecar.go index c9d264bb..811dab74 100644 --- a/internal/node/sidecar.go +++ b/internal/node/sidecar.go @@ -692,7 +692,8 @@ func createEgressProxy(node *SamNode) http.Handler { req.URL.RawPath = "" logger.Debugf("[Proxy] Rewriting URL to libp2p://%s%s", req.URL.Host, req.URL.Path) }, - Transport: transport, + Transport: transport, + ModifyResponse: rewriteA2AAgentCard, } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -708,6 +709,11 @@ func createEgressProxy(node *SamNode) http.Handler { return } + r, ok := a2aEgressHook(node, w, r) + if !ok { + return + } + r.Header.Set(api.HeaderSamBiscuit, base64.StdEncoding.EncodeToString(biscuitBytes)) // Forwarded, not stripped: the agent claim is what lets the peer at the From b29eef04e253a5e9e991a66f35ec332efde8b02d Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 09:39:39 +0000 Subject: [PATCH 04/16] node: cover a2a egress edge cases in unit tests --- internal/node/a2a_service_test.go | 53 +++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index 2418b3d9..5a580117 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -172,3 +172,56 @@ func TestRewriteA2AAgentCardNoopWithoutTag(t *testing.T) { t.Fatalf("untagged response was modified: %s", body) } } + +func TestA2AEgressHookInvalidPeerID(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/not-a-peer/a2a/agent/", nil) + req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") + _, ok := a2aEgressHook(nil, rec, req) + if ok { + t.Fatal("invalid peer ID must be refused") + } + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", rec.Code) + } +} + +func TestRewriteA2AAgentCardSkipsNon200(t *testing.T) { + orig := `{"name":"T","url":"http://localhost:9999"}` + base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) + resp := &http.Response{ + StatusCode: http.StatusNotFound, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader(orig)), + Request: req, + } + if err := rewriteA2AAgentCard(resp); err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + if string(body) != orig { + t.Fatalf("non-200 response was modified: %s", body) + } +} + +func TestRewriteA2AAgentCardSkipsContentEncoded(t *testing.T) { + orig := `{"name":"T","url":"http://localhost:9999"}` + base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Encoding": []string{"gzip"}}, + Body: io.NopCloser(strings.NewReader(orig)), + Request: req, + } + if err := rewriteA2AAgentCard(resp); err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + if string(body) != orig { + t.Fatalf("content-encoded response was modified: %s", body) + } +} From 2f21b7af02093057b01f32a46574555bd4e499c7 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 09:57:15 +0000 Subject: [PATCH 05/16] docs: document a2a service routing and the egress labels gate --- site/content/docs/user/node-configuration.md | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/site/content/docs/user/node-configuration.md b/site/content/docs/user/node-configuration.md index dd0f41da..fb74fd8b 100644 --- a/site/content/docs/user/node-configuration.md +++ b/site/content/docs/user/node-configuration.md @@ -60,7 +60,7 @@ The `services` array allows you to register endpoints that remote peers in the S | `description` | A human-readable description published to the mesh discovery catalogue. | | `command` | *(For MCP)* The executable command array to spawn as a local subprocess (e.g. `["node", "index.js"]`). | | `env` | *(For MCP)* Key-value environment variables passed to the subprocess. | -| `target_url` | *(For HTTP/Inference)* The upstream local URL to proxy traffic to. | +| `target_url` | *(For HTTP/Inference/A2A)* The upstream local URL to proxy traffic to. | ### Inference Service Path Standards & Proxy Routing @@ -69,6 +69,13 @@ When configuring `target_url` for `type: inference` services (e.g. Ollama, vLLM, * **OpenAI Facade Access**: Clients connecting via the node's local OpenAI Facade (`http://localhost:8080/v1`) request paths like `/v1/chat/completions`. SAM automatically proxies these to the backend's root URL. * **Raw Proxy Access**: If bypassing the Facade and making requests directly via the local egress proxy (`/sam/{peer}/inference/{service}`), the request path must include the explicit `/v1` namespace suffix (e.g. `http://localhost:8080/sam/{peer}/inference/{service}/v1/chat/completions`). +### A2A Service Routing + +When configuring `type: a2a` services (Agent2Agent protocol agents): +* **URL backends only**: register the agent's local HTTP endpoint as `target_url`. `command` backends are rejected. +* **Raw Proxy Access**: remote peers reach the agent at `http://localhost:8080/sam/{peer}/a2a/{service}/...`. The agent card served at `.../.well-known/agent-card.json` is rewritten in transit so its interface URLs point back at this mesh path; transports the mesh cannot carry (gRPC) are dropped and streaming is advertised off. +* **Label-gated egress**: setting `X-Sam-Required-Labels: key=value[,key=value]` on a raw a2a request makes the caller's node verify the provider's control-plane-attested labels and refuse fail-closed (HTTP 403) before any data leaves the node. The header is stripped before forwarding. + --- ## 3. Defining Local Security (Target Attenuation) From 4a0979036220239fa6b431176a0091b043db7f07 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 20:39:09 +0000 Subject: [PATCH 06/16] tests: cover the a2a mesh CUJ end to end --- tests/integration/a2a_test.go | 231 ++++++++++++++++++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 tests/integration/a2a_test.go diff --git a/tests/integration/a2a_test.go b/tests/integration/a2a_test.go new file mode 100644 index 00000000..5779b0fb --- /dev/null +++ b/tests/integration/a2a_test.go @@ -0,0 +1,231 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package integration_test + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/encoding/protojson" + + "github.com/google/sam/api" +) + +// TestA2ACUJ covers the "A2A agent behind the mesh" CUJ: node A (attested +// region=eu) hosts an a2a service; node B's raw egress proxy serves it with +// a rewritten agent card, admits a region=eu-labelled request, and refuses a +// region=us-east-1 request fail-closed before any payload leaves node B. +func TestA2ACUJ(t *testing.T) { + nodeBin := buildBinary(t, "./cmd/sam-node") + _, hubAddr := startMockRouter(t) + + homeA := t.TempDir() + homeB := t.TempDir() + apiToken := "test-token" + + t.Log("Starting Node A (provider, region=eu)...") + _ = startBackgroundNode(t, nodeBin, hubAddr, homeA, + "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", + "--listen", "/ip4/127.0.0.1/tcp/0", + "--discovery-interval", "100ms", + "--labels", "region=eu", + ) + t.Log("Starting Node B (consumer)...") + _ = startBackgroundNode(t, nodeBin, hubAddr, homeB, + "--listen", "/ip4/127.0.0.1/udp/0/quic-v1", + "--listen", "/ip4/127.0.0.1/tcp/0", + "--discovery-interval", "100ms", + ) + + apiAddrA := waitForMCPAddr(t, filepath.Join(homeA, "node.log")) + apiAddrB := waitForMCPAddr(t, filepath.Join(homeB, "node.log")) + waitForAPI(t, apiAddrA) + waitForAPI(t, apiAddrB) + + addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) + callMCP(t, apiAddrB, "connect_peer", map[string]any{"peer_addr": addrA}) + waitForDHTPeers(t, apiAddrA) + + idx := strings.LastIndex(addrA, "/p2p/") + if idx < 0 { + t.Fatalf("no /p2p/ component in peer addr %q", addrA) + } + peerA := addrA[idx+len("/p2p/"):] + + // Fake A2A agent on node A's side: serves its card and echoes message/send. + var sendCount atomic.Int32 + var sawLabelsHeader atomic.Bool + agent := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get(api.HeaderSamRequiredLabels) != "" { + sawLabelsHeader.Store(true) + } + switch { + case r.Method == http.MethodGet && r.URL.Path == "/.well-known/agent-card.json": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"name":"echo-agent","url":"http://localhost:9999",` + + `"preferredTransport":"JSONRPC",` + + `"additionalInterfaces":[{"url":"http://localhost:9999","transport":"JSONRPC"},` + + `{"url":"localhost:50051","transport":"GRPC"}],` + + `"capabilities":{"streaming":true}}`)) + case r.Method == http.MethodPost: + sendCount.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"kind":"message",` + + `"messageId":"m1","role":"agent","parts":[{"kind":"text","text":"echo from eu"}]}}`)) + default: + http.Error(w, "not found", http.StatusNotFound) + } + })) + defer agent.Close() + + registerA2AService(t, apiAddrA, apiToken, "echo-agent", agent.URL) + + meshBase := "http://" + apiAddrB + "/sam/" + peerA + "/a2a/echo-agent" + + // CUJ step 1: the agent card comes back rewritten for mesh use. Poll: + // the first fetch can race connectivity establishment. + deadline := time.Now().Add(30 * time.Second) + var cardBody []byte + for { + req, _ := http.NewRequest("GET", meshBase+"/.well-known/agent-card.json", nil) + req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) + resp, err := http.DefaultClient.Do(req) + if err == nil { + cardBody, _ = io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + break + } + } + if time.Now().After(deadline) { + t.Fatalf("timeout fetching agent card, last body: %s", string(cardBody)) + } + time.Sleep(200 * time.Millisecond) + } + var card struct { + URL string `json:"url"` + PreferredTransport string `json:"preferredTransport"` + AdditionalInterfaces []struct { + URL string `json:"url"` + Transport string `json:"transport"` + } `json:"additionalInterfaces"` + Capabilities struct { + Streaming bool `json:"streaming"` + } `json:"capabilities"` + } + if err := json.Unmarshal(cardBody, &card); err != nil { + t.Fatalf("invalid card: %v, body: %s", err, string(cardBody)) + } + if card.URL != meshBase { + t.Errorf("card url = %q, want mesh base %q", card.URL, meshBase) + } + if len(card.AdditionalInterfaces) != 1 || card.AdditionalInterfaces[0].URL != meshBase { + t.Errorf("interfaces not rewritten / gRPC not dropped: %s", string(cardBody)) + } + if card.Capabilities.Streaming { + t.Error("streaming must be advertised off through the mesh") + } + + // CUJ step 2: message/send constrained to region=eu is admitted. + sendBody := `{"jsonrpc":"2.0","id":1,"method":"message/send","params":{"message":` + + `{"kind":"message","messageId":"c1","role":"user","parts":[{"kind":"text","text":"hi"}]}}}` + deadline = time.Now().Add(30 * time.Second) + for { + req, _ := http.NewRequest("POST", meshBase+"/", strings.NewReader(sendBody)) + req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("labelled message/send failed: %v", err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + if !strings.Contains(string(body), "echo from eu") { + t.Fatalf("unexpected message/send response: %s", string(body)) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("labelled message/send status: %d, body: %s", resp.StatusCode, string(body)) + } + time.Sleep(200 * time.Millisecond) + } + + // CUJ step 3: a mismatched label refuses fail-closed BEFORE egress — + // the agent backend must never see the request. + before := sendCount.Load() + req, _ := http.NewRequest("POST", meshBase+"/", strings.NewReader(sendBody)) + req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(api.HeaderSamRequiredLabels, "region=us-east-1") + respUS, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("mismatched-label message/send failed: %v", err) + } + usBody, _ := io.ReadAll(respUS.Body) + _ = respUS.Body.Close() + if respUS.StatusCode != http.StatusForbidden { + t.Fatalf("mismatched label must fail closed with 403: got %d, body: %s", respUS.StatusCode, string(usBody)) + } + if sendCount.Load() != before { + t.Fatal("payload reached the agent despite label refusal") + } + + // Zero-trust invariant: the labels header never crosses the mesh. + if sawLabelsHeader.Load() { + t.Errorf("%s header leaked to the agent backend", api.HeaderSamRequiredLabels) + } + + t.Log("A2A CUJ test passed.") +} + +func registerA2AService(t *testing.T, apiAddr, token, serviceName, targetURL string) { + t.Helper() + reqData := &api.RegisterServiceRequest{ + Service: &api.ServiceInfo{ + Type: api.ServiceType_SERVICE_TYPE_A2A, + Name: serviceName, + Description: "test a2a agent", + }, + Backend: &api.RegisterServiceRequest_TargetUrl{TargetUrl: targetURL}, + } + body, err := protojson.Marshal(reqData) + if err != nil { + t.Fatal(err) + } + req, _ := http.NewRequest("POST", "http://"+apiAddr+"/sam/service/register", bytes.NewBuffer(body)) + req.Header.Set(api.HeaderSamAuthentication, "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("Failed to register a2a service: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(resp.Body) + t.Fatalf("Register a2a service failed with status: %d, body: %s", resp.StatusCode, string(bodyBytes)) + } +} From 830d0bad74ea42758520a7673fa88e6dd4ee6d05 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sat, 29 Aug 2026 20:46:13 +0000 Subject: [PATCH 07/16] node: gate a2a egress case-insensitively to match ingress --- internal/node/a2a_service.go | 2 +- internal/node/a2a_service_test.go | 16 ++++++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index f5429b5a..3045739d 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -58,7 +58,7 @@ type a2aCardBaseURL struct{} // On refusal it writes the HTTP error itself and returns ok=false. func a2aEgressHook(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { parts := strings.SplitN(r.URL.Path, "/", 6) - if len(parts) < 5 || parts[3] != api.ServiceTypeStringA2A { + if len(parts) < 5 || !strings.EqualFold(parts[3], api.ServiceTypeStringA2A) { return r, true } if labelsHeader := r.Header.Get(api.HeaderSamRequiredLabels); labelsHeader != "" { diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index 5a580117..e5917cad 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -225,3 +225,19 @@ func TestRewriteA2AAgentCardSkipsContentEncoded(t *testing.T) { t.Fatalf("content-encoded response was modified: %s", body) } } + +func TestA2AEgressHookUppercaseTypeIsGated(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/sam/not-a-peer/A2A/agent/", nil) + req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") + _, ok := a2aEgressHook(nil, rec, req) + if ok { + t.Fatal("uppercase A2A path must not bypass the labels gate") + } + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (invalid peer reached after gate engaged)", rec.Code) + } + if req.Header.Get(api.HeaderSamRequiredLabels) != "" { + t.Fatal("labels header must be stripped on a2a paths regardless of case") + } +} From 84fc8ea333a038833781f13a6af340be4fd90b43 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sun, 30 Aug 2026 16:54:51 +0000 Subject: [PATCH 08/16] node: dispatch egress middleware by service type instead of hardcoding a2a --- internal/node/a2a_service.go | 23 ++++++------ internal/node/a2a_service_test.go | 10 +++--- internal/node/sidecar.go | 59 +++++++++++++++++++++++++++++-- 3 files changed, 75 insertions(+), 17 deletions(-) diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index 3045739d..66a88f0a 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -49,18 +49,21 @@ func (s *A2AService) Init(ctx context.Context) error { return nil } +func init() { + registerEgressMiddleware(api.ServiceTypeStringA2A, egressMiddleware{ + gateRequest: a2aEgressGate, + modifyResponse: rewriteA2AAgentCard, + }) +} + // a2aCardBaseURL is the context key carrying the caller-facing mesh base URL -// of an agent-card fetch, set by a2aEgressHook and consumed by the rewrite. +// of an agent-card fetch, set by a2aEgressGate and consumed by the rewrite. type a2aCardBaseURL struct{} -// a2aEgressHook runs the caller-side A2A checks on a raw egress request: +// a2aEgressGate runs the caller-side A2A checks on a raw egress request: // the fail-closed labels gate and tagging agent-card fetches for rewrite. // On refusal it writes the HTTP error itself and returns ok=false. -func a2aEgressHook(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { - parts := strings.SplitN(r.URL.Path, "/", 6) - if len(parts) < 5 || !strings.EqualFold(parts[3], api.ServiceTypeStringA2A) { - return r, true - } +func a2aEgressGate(node *SamNode, w http.ResponseWriter, r *http.Request, route egressRoute) (*http.Request, bool) { if labelsHeader := r.Header.Get(api.HeaderSamRequiredLabels); labelsHeader != "" { r.Header.Del(api.HeaderSamRequiredLabels) required, err := parseRequiredLabels(labelsHeader) @@ -68,19 +71,19 @@ func a2aEgressHook(node *SamNode, w http.ResponseWriter, r *http.Request) (*http http.Error(w, fmt.Sprintf("Invalid %s header: %v", api.HeaderSamRequiredLabels, err), http.StatusBadRequest) return r, false } - pid, err := peer.Decode(parts[2]) + pid, err := peer.Decode(route.peerID) if err != nil { http.Error(w, "Invalid peer ID", http.StatusBadRequest) return r, false } if err := node.VerifyPeerLabels(r.Context(), pid, required); err != nil { - logger.Warnf("[A2A] label gate refused egress to %s: %v", parts[2], err) + logger.Warnf("[A2A] label gate refused egress to %s: %v", route.peerID, err) http.Error(w, "Required labels not attested by provider", http.StatusForbidden) return r, false } } if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/.well-known/agent-card.json") { - base := fmt.Sprintf("http://%s/sam/%s/%s/%s", r.Host, parts[2], parts[3], parts[4]) + base := fmt.Sprintf("http://%s/sam/%s/%s/%s", r.Host, route.peerID, route.serviceType, route.serviceName) r = r.WithContext(context.WithValue(r.Context(), a2aCardBaseURL{}, base)) } return r, true diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index e5917cad..9a4c38f5 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -67,7 +67,7 @@ func TestA2AEgressHookNonA2APassthrough(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/mcp/svc/foo", nil) req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") - _, ok := a2aEgressHook(nil, rec, req) + _, ok := applyEgressMiddleware(nil, rec, req) if !ok { t.Fatal("non-a2a path must pass through") } @@ -80,7 +80,7 @@ func TestA2AEgressHookMalformedLabels(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest("POST", "/sam/12D3KooWpeer/a2a/agent/", nil) req.Header.Set(api.HeaderSamRequiredLabels, "not-a-label") - _, ok := a2aEgressHook(nil, rec, req) + _, ok := applyEgressMiddleware(nil, rec, req) if ok { t.Fatal("malformed labels must be refused") } @@ -93,7 +93,7 @@ func TestA2AEgressHookTagsCardFetch(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) req.Host = "127.0.0.1:8080" - r2, ok := a2aEgressHook(nil, rec, req) + r2, ok := applyEgressMiddleware(nil, rec, req) if !ok { t.Fatal("card fetch must pass through") } @@ -177,7 +177,7 @@ func TestA2AEgressHookInvalidPeerID(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/not-a-peer/a2a/agent/", nil) req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") - _, ok := a2aEgressHook(nil, rec, req) + _, ok := applyEgressMiddleware(nil, rec, req) if ok { t.Fatal("invalid peer ID must be refused") } @@ -230,7 +230,7 @@ func TestA2AEgressHookUppercaseTypeIsGated(t *testing.T) { rec := httptest.NewRecorder() req := httptest.NewRequest("POST", "/sam/not-a-peer/A2A/agent/", nil) req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") - _, ok := a2aEgressHook(nil, rec, req) + _, ok := applyEgressMiddleware(nil, rec, req) if ok { t.Fatal("uppercase A2A path must not bypass the labels gate") } diff --git a/internal/node/sidecar.go b/internal/node/sidecar.go index 811dab74..4b6992ab 100644 --- a/internal/node/sidecar.go +++ b/internal/node/sidecar.go @@ -653,6 +653,61 @@ func handleDiscoverService(node *SamNode, w http.ResponseWriter, r *http.Request } } +// egressMiddleware lets a service type hook raw egress traffic without the +// proxy knowing the type exists. Types self-register from init(). +type egressMiddleware struct { + // gateRequest may refuse the request (writing the HTTP error itself, + // returning false) or return a derived request to forward. + gateRequest func(node *SamNode, w http.ResponseWriter, r *http.Request, route egressRoute) (*http.Request, bool) + // modifyResponse edits the proxied response; nil means no hook. + modifyResponse func(*http.Response) error +} + +// egressRoute is the parsed /sam/{peer}/{type}/{svc}/{upstream} egress path, +// with the service type lowercased to match how the remote ingress parses it. +type egressRoute struct { + peerID string + serviceType string + serviceName string + upstreamPath string +} + +// Keyed by lowercase service-type string; written only during init(). +var egressMiddlewares = map[string]egressMiddleware{} + +func registerEgressMiddleware(serviceType string, mw egressMiddleware) { + egressMiddlewares[strings.ToLower(serviceType)] = mw +} + +// applyEgressMiddleware routes a raw egress request through the middleware +// registered for its service type, if any. +func applyEgressMiddleware(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { + parts := strings.SplitN(r.URL.Path, "/", 6) + if len(parts) < 5 { + return r, true + } + route := egressRoute{peerID: parts[2], serviceType: strings.ToLower(parts[3]), serviceName: parts[4]} + if len(parts) > 5 { + route.upstreamPath = parts[5] + } + mw, ok := egressMiddlewares[route.serviceType] + if !ok || mw.gateRequest == nil { + return r, true + } + return mw.gateRequest(node, w, r, route) +} + +// egressModifyResponse dispatches to the middleware of the service type in +// the rewritten path (/{type}/{svc}/...); types without a hook pass through. +func egressModifyResponse(resp *http.Response) error { + parts := strings.SplitN(strings.TrimPrefix(resp.Request.URL.Path, "/"), "/", 2) + mw, ok := egressMiddlewares[strings.ToLower(parts[0])] + if !ok || mw.modifyResponse == nil { + return nil + } + return mw.modifyResponse(resp) +} + func createEgressProxy(node *SamNode) http.Handler { transport := libp2phttp.NewTransport(node.Host) @@ -693,7 +748,7 @@ func createEgressProxy(node *SamNode) http.Handler { logger.Debugf("[Proxy] Rewriting URL to libp2p://%s%s", req.URL.Host, req.URL.Path) }, Transport: transport, - ModifyResponse: rewriteA2AAgentCard, + ModifyResponse: egressModifyResponse, } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -709,7 +764,7 @@ func createEgressProxy(node *SamNode) http.Handler { return } - r, ok := a2aEgressHook(node, w, r) + r, ok := applyEgressMiddleware(node, w, r) if !ok { return } From e403974b7676390e6f8390a5c8df1f40f91f1abe Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sun, 30 Aug 2026 21:00:55 +0000 Subject: [PATCH 09/16] refactor for readibility --- internal/node/a2a_service.go | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index 66a88f0a..f3282907 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -28,6 +28,13 @@ import ( "github.com/libp2p/go-libp2p/core/peer" ) +func init() { + registerEgressMiddleware(api.ServiceTypeStringA2A, egressMiddleware{ + gateRequest: a2aEgressGate, + modifyResponse: rewriteA2AAgentCard, + }) +} + // A2AService proxies Agent2Agent (A2A) JSON-RPC/REST traffic to a local // agent process. URL backends only: a command backend would wire the A2A // route to an MCP stdio bridge no A2A client can talk to. @@ -49,13 +56,6 @@ func (s *A2AService) Init(ctx context.Context) error { return nil } -func init() { - registerEgressMiddleware(api.ServiceTypeStringA2A, egressMiddleware{ - gateRequest: a2aEgressGate, - modifyResponse: rewriteA2AAgentCard, - }) -} - // a2aCardBaseURL is the context key carrying the caller-facing mesh base URL // of an agent-card fetch, set by a2aEgressGate and consumed by the rewrite. type a2aCardBaseURL struct{} @@ -91,7 +91,7 @@ func a2aEgressGate(node *SamNode, w http.ResponseWriter, r *http.Request, route // rewriteA2AAgentCard makes a proxied agent card usable by stock A2A clients: // interface URLs point back at the mesh path, transports the mesh cannot -// carry are dropped, and streaming is advertised off until verified. +// carry (gRPC) are dropped, and streaming is advertised off until verified. func rewriteA2AAgentCard(resp *http.Response) error { base, ok := resp.Request.Context().Value(a2aCardBaseURL{}).(string) if !ok || resp.StatusCode != http.StatusOK { From 226740b24e998b1f2bf703e15c9c39ec21755b3a Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Sun, 30 Aug 2026 21:24:09 +0000 Subject: [PATCH 10/16] node: move parseRequiredLabels next to the label gate --- internal/node/labels_gate.go | 36 +++++++++++++++++++++++++++++++ internal/node/openai_scorer.go | 39 ---------------------------------- 2 files changed, 36 insertions(+), 39 deletions(-) diff --git a/internal/node/labels_gate.go b/internal/node/labels_gate.go index 33d78478..6adcbe92 100644 --- a/internal/node/labels_gate.go +++ b/internal/node/labels_gate.go @@ -19,6 +19,7 @@ import ( "crypto/ed25519" "fmt" "sort" + "strings" "time" "github.com/google/sam/api" @@ -28,6 +29,41 @@ import ( "google.golang.org/protobuf/proto" ) +// parseRequiredLabels splits the X-Sam-Required-Labels header value +// (comma-separated "key=value" pairs) into a label map; any malformed entry +// rejects the whole request. +func parseRequiredLabels(h string) (map[string]string, error) { + if h == "" { + return nil, nil + } + var out map[string]string + for _, part := range strings.Split(h, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + k, v, ok := strings.Cut(part, "=") + if !ok { + return nil, fmt.Errorf("invalid label %q: expected key=value", part) + } + k, v = strings.TrimSpace(k), strings.TrimSpace(v) + if err := api.ValidateLabelKey(k); err != nil { + return nil, err + } + if err := api.ValidateLabelValue(v); err != nil { + return nil, err + } + if out == nil { + out = make(map[string]string) + } + if _, exists := out[k]; exists { + return nil, fmt.Errorf("duplicate label key %q", k) + } + out[k] = v + } + return out, nil +} + // The label gate is the consumer-side enforcement point for label // requirements: gossip labels only rank providers, this gate verifies the // provider's control-plane-attested label() facts (api.FactLabel) before diff --git a/internal/node/openai_scorer.go b/internal/node/openai_scorer.go index 5fd33cea..b7a34561 100644 --- a/internal/node/openai_scorer.go +++ b/internal/node/openai_scorer.go @@ -15,13 +15,9 @@ package node import ( - "fmt" "net/http" "sort" - "strings" "time" - - "github.com/google/sam/api" ) // providerBackoff is how long a provider is skipped after a retryable failure. @@ -37,41 +33,6 @@ const ( reasonAttemptsExceeded = "attempts_exceeded" ) -// parseRequiredLabels splits the X-Sam-Required-Labels header value -// (comma-separated "key=value" pairs) into a label map; any malformed entry -// rejects the whole request. -func parseRequiredLabels(h string) (map[string]string, error) { - if h == "" { - return nil, nil - } - var out map[string]string - for _, part := range strings.Split(h, ",") { - part = strings.TrimSpace(part) - if part == "" { - continue - } - k, v, ok := strings.Cut(part, "=") - if !ok { - return nil, fmt.Errorf("invalid label %q: expected key=value", part) - } - k, v = strings.TrimSpace(k), strings.TrimSpace(v) - if err := api.ValidateLabelKey(k); err != nil { - return nil, err - } - if err := api.ValidateLabelValue(v); err != nil { - return nil, err - } - if out == nil { - out = make(map[string]string) - } - if _, exists := out[k]; exists { - return nil, fmt.Errorf("duplicate label key %q", k) - } - out[k] = v - } - return out, nil -} - // labelsAllowed reports whether a provider's claimed labels satisfy any // required key=value pair (exact match). func labelsAllowed(required, claimed map[string]string) bool { From 20e5799acba052657b5d52dc50b0f1f665ee4e32 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 1 Sep 2026 13:02:53 +0000 Subject: [PATCH 11/16] examples: add chat-a2a, a Gemini-backed A2A agent with a chat REPL for the kind mesh --- development/examples/chat-a2a/Dockerfile | 10 ++ development/examples/chat-a2a/README.md | 70 ++++++++++++ development/examples/chat-a2a/agent.py | 108 ++++++++++++++++++ development/examples/chat-a2a/chat.py | 44 +++++++ .../examples/chat-a2a/requirements.txt | 6 + .../examples/chat-a2a/sam-node-config.yaml | 8 ++ 6 files changed, 246 insertions(+) create mode 100644 development/examples/chat-a2a/Dockerfile create mode 100644 development/examples/chat-a2a/README.md create mode 100644 development/examples/chat-a2a/agent.py create mode 100644 development/examples/chat-a2a/chat.py create mode 100644 development/examples/chat-a2a/requirements.txt create mode 100644 development/examples/chat-a2a/sam-node-config.yaml diff --git a/development/examples/chat-a2a/Dockerfile b/development/examples/chat-a2a/Dockerfile new file mode 100644 index 00000000..2e474c2b --- /dev/null +++ b/development/examples/chat-a2a/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.12-slim + +WORKDIR /srv +COPY requirements.txt /srv/ +RUN pip install --no-cache-dir -r requirements.txt +COPY agent.py /srv/ + +ENV GEMINI_API_KEY= + +CMD ["python3", "/srv/agent.py"] diff --git a/development/examples/chat-a2a/README.md b/development/examples/chat-a2a/README.md new file mode 100644 index 00000000..f98cdc77 --- /dev/null +++ b/development/examples/chat-a2a/README.md @@ -0,0 +1,70 @@ +# chat-a2a + +A Gemini-backed A2A agent plus a tiny chat REPL, exercising the mesh's a2a +support end to end: agent-card fetch with caller-side rewrite, `message/send` +routing over libp2p, and `contextId` continuity across turns. + +## 1. Set your Gemini key + +Edit `Dockerfile` and replace `` in `ENV GEMINI_API_KEY=` +(same pattern as `gemini-buddy-mcp`). Optionally override the model with +`GEMINI_MODEL` (default `models/gemini-3.6-flash`). + +## 2. Host the agent on a mesh node + +Assign it in `development/kind/mesh-config.yaml`: + +```yaml +node-a: +node-b: chat-a2a +``` + +Then bring the mesh up: + +```sh +make build +make kind-up +``` + +## 3. Enroll a local caller node + +```sh +./development/kind/run-local-node.sh +``` + +Sidecar API lands on `127.0.0.1:9099` with token `devtoken`. + +## 4. Find the provider peer + +```sh +./bin/mcp-client -url http://127.0.0.1:9099/mcp -token devtoken \ + -tool discover_remote_services -args '{"type":"a2a","name":"chat"}' +``` + +Note the peer ID and export it: `export PEER=`. Discovery is +gossip-fed; retry for a few seconds after startup if the list comes back empty. + +## 5. See the card rewrite + +```sh +curl -s -H 'X-Sam-Authentication: Bearer devtoken' \ + "http://127.0.0.1:9099/sam/$PEER/a2a/chat/.well-known/agent-card.json" | jq +``` + +The interface URLs point back at this mesh path (not the agent's own address) +and `capabilities.streaming` is forced to `false` — that rewrite is what lets +a stock A2A client work against the mesh unmodified. + +## 6. Chat + +Requires [`uv`](https://docs.astral.sh/uv/). + +```sh +uv run --with-requirements requirements.txt chat.py "http://127.0.0.1:9099/sam/$PEER/a2a/chat" +``` + +Tell the agent your name, then ask for it back a couple of turns later: the +client carries the `contextId` the server minted on the first reply, and the +agent keeps one Gemini chat session per context, so the answer proves the +conversation survived the mesh hop. Each turn is still its own short-lived +A2A task — `taskId` changes every turn, `contextId` is what persists. diff --git a/development/examples/chat-a2a/agent.py b/development/examples/chat-a2a/agent.py new file mode 100644 index 00000000..2f4fd4c7 --- /dev/null +++ b/development/examples/chat-a2a/agent.py @@ -0,0 +1,108 @@ +"""Gemini-backed A2A chat agent hosted by a node in the local dev mesh.""" +import os +import time +import uuid + +import uvicorn +from a2a.server.agent_execution.agent_executor import AgentExecutor +from a2a.server.agent_execution.context import RequestContext +from a2a.server.events.event_queue import EventQueue +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes +from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore +from a2a.types import ( + AgentCapabilities, + AgentCard, + AgentInterface, + AgentSkill, + Message, + Part, + Role, +) +from google import genai +from google.genai import types +from starlette.applications import Starlette + +PORT = 7777 +MODEL = os.environ.get("GEMINI_MODEL", "models/gemini-3.6-flash") + +class ChatExecutor(AgentExecutor): + """One Gemini chat session per A2A contextId; the session carries the history.""" + + def __init__(self): + self.gemini = genai.Client() + self.chats = {} + + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + chat = self.chats.get(context.context_id) + if chat is None: + # Gemini 3 Flash defaults to thinking_level=high, which dominates latency. + chat = self.gemini.aio.chats.create( + model=MODEL, + config=types.GenerateContentConfig( + thinking_config=types.ThinkingConfig(thinking_level="minimal") + ), + ) + self.chats[context.context_id] = chat + started = time.monotonic() + reply = await chat.send_message(context.get_user_input()) + print( + f"[chat] context={context.context_id} gemini took " + f"{time.monotonic() - started:.1f}s usage={reply.usage_metadata}", + flush=True, + ) + await event_queue.enqueue_event( + Message( + role=Role.ROLE_AGENT, + message_id=str(uuid.uuid4()), + parts=[Part(text=reply.text or "")], + context_id=context.context_id, + task_id=context.task_id, + ) + ) + + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: + pass + + +agent_card = AgentCard( + name="chat", + description="Gemini-backed conversational agent; remembers the conversation per contextId", + version="0.1.0", + capabilities=AgentCapabilities(streaming=False), + default_input_modes=["text"], + default_output_modes=["text"], + skills=[ + AgentSkill( + id="chat", + name="chat", + description="Multi-turn small talk", + tags=["chat"], + examples=["hi, my name is Ada", "what is my name?"], + ) + ], + supported_interfaces=[ + AgentInterface( + protocol_binding="JSONRPC", + protocol_version="1.0", + url=f"http://127.0.0.1:{PORT}/", + ) + ], +) + +handler = DefaultRequestHandler( + agent_executor=ChatExecutor(), + task_store=InMemoryTaskStore(), + agent_card=agent_card, +) +# Starlette over FastAPI: the SDK generates the routes, so FastAPI would add nothing. +# JSON-RPC at "/": the mesh card rewrite drops URL subpaths, so clients land on the root. +app = Starlette( + routes=[ + *create_jsonrpc_routes(request_handler=handler, rpc_url="/"), + *create_agent_card_routes(agent_card=agent_card), + ] +) + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=PORT) diff --git a/development/examples/chat-a2a/chat.py b/development/examples/chat-a2a/chat.py new file mode 100644 index 00000000..f23fa2dc --- /dev/null +++ b/development/examples/chat-a2a/chat.py @@ -0,0 +1,44 @@ +"""Minimal A2A chat REPL: resolves the agent card through the mesh and talks to it.""" +import asyncio +import os +import sys +import uuid + +import httpx +from a2a.client import A2ACardResolver, ClientConfig, create_client +from a2a.helpers import get_message_text +from a2a.types import Message, Part, Role, SendMessageRequest + + +async def main(url: str) -> None: + token = os.environ.get("SAM_API_TOKEN", "devtoken") + async with httpx.AsyncClient(timeout=120, headers={"X-Sam-Authentication": f"Bearer {token}"}) as http: + card = await A2ACardResolver(http, url).get_agent_card() + print(f"{card.name}: {card.description}") + for iface in card.supported_interfaces: + print(f" {iface.protocol_binding} -> {iface.url}") + client = await create_client(card, client_config=ClientConfig(httpx_client=http)) + context_id = None + while True: + try: + text = input("you> ") + except EOFError: + return + if not text.strip(): + continue + message = Message( + role=Role.ROLE_USER, + message_id=str(uuid.uuid4()), + parts=[Part(text=text)], + context_id=context_id, + ) + async for event in client.send_message(SendMessageRequest(message=message)): + if event.HasField("message"): + context_id = event.message.context_id or context_id + print(f"agent> {get_message_text(event.message)}") + + +if __name__ == "__main__": + if len(sys.argv) != 2: + sys.exit("usage: chat.py http://127.0.0.1:9099/sam//a2a/chat") + asyncio.run(main(sys.argv[1])) diff --git a/development/examples/chat-a2a/requirements.txt b/development/examples/chat-a2a/requirements.txt new file mode 100644 index 00000000..ee9c850d --- /dev/null +++ b/development/examples/chat-a2a/requirements.txt @@ -0,0 +1,6 @@ +a2a-sdk>=1.0 +google-genai>=1.0 +httpx>=0.27 +sse-starlette>=2.0 +starlette>=0.40 +uvicorn>=0.30 diff --git a/development/examples/chat-a2a/sam-node-config.yaml b/development/examples/chat-a2a/sam-node-config.yaml new file mode 100644 index 00000000..910a4901 --- /dev/null +++ b/development/examples/chat-a2a/sam-node-config.yaml @@ -0,0 +1,8 @@ +version: "v1alpha1" +attenuation: + policies: [] +services: + - type: "a2a" + name: "chat" + description: "Gemini-backed conversational A2A agent" + target_url: "http://127.0.0.1:7777" From 8529ef8005d22a9c2ee58525078e2c3ed7daef35 Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 1 Sep 2026 16:04:03 +0200 Subject: [PATCH 12/16] Use gemini-3.5-flash-lite for the agent --- development/examples/chat-a2a/agent.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/development/examples/chat-a2a/agent.py b/development/examples/chat-a2a/agent.py index 2f4fd4c7..ec4ec8a6 100644 --- a/development/examples/chat-a2a/agent.py +++ b/development/examples/chat-a2a/agent.py @@ -24,7 +24,7 @@ from starlette.applications import Starlette PORT = 7777 -MODEL = os.environ.get("GEMINI_MODEL", "models/gemini-3.6-flash") +MODEL = os.environ.get("GEMINI_MODEL", "models/gemini-3.5-flash-lite") class ChatExecutor(AgentExecutor): """One Gemini chat session per A2A contextId; the session carries the history.""" From 69cbe36a4d66b01681ef88c826096e4ff645ef3b Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 1 Sep 2026 20:33:46 +0000 Subject: [PATCH 13/16] docs: add the chat-a2a use case page and wire GEMINI_MODEL through the example --- development/examples/chat-a2a/Dockerfile | 1 + development/examples/chat-a2a/README.md | 3 +- site/content/docs/use-cases/chat-a2a.md | 147 +++++++++++++++++++ site/content/docs/user/node-configuration.md | 1 + 4 files changed, 151 insertions(+), 1 deletion(-) create mode 100644 site/content/docs/use-cases/chat-a2a.md diff --git a/development/examples/chat-a2a/Dockerfile b/development/examples/chat-a2a/Dockerfile index 2e474c2b..57f1fdb2 100644 --- a/development/examples/chat-a2a/Dockerfile +++ b/development/examples/chat-a2a/Dockerfile @@ -6,5 +6,6 @@ RUN pip install --no-cache-dir -r requirements.txt COPY agent.py /srv/ ENV GEMINI_API_KEY= +ENV GEMINI_MODEL=models/gemini-3.5-flash-lite CMD ["python3", "/srv/agent.py"] diff --git a/development/examples/chat-a2a/README.md b/development/examples/chat-a2a/README.md index f98cdc77..5d989f61 100644 --- a/development/examples/chat-a2a/README.md +++ b/development/examples/chat-a2a/README.md @@ -8,7 +8,7 @@ routing over libp2p, and `contextId` continuity across turns. Edit `Dockerfile` and replace `` in `ENV GEMINI_API_KEY=` (same pattern as `gemini-buddy-mcp`). Optionally override the model with -`GEMINI_MODEL` (default `models/gemini-3.6-flash`). +`GEMINI_MODEL` (default `models/gemini-3.5-flash-lite`). ## 2. Host the agent on a mesh node @@ -60,6 +60,7 @@ a stock A2A client work against the mesh unmodified. Requires [`uv`](https://docs.astral.sh/uv/). ```sh +cd development/examples/chat-a2a uv run --with-requirements requirements.txt chat.py "http://127.0.0.1:9099/sam/$PEER/a2a/chat" ``` diff --git a/site/content/docs/use-cases/chat-a2a.md b/site/content/docs/use-cases/chat-a2a.md new file mode 100644 index 00000000..accb519e --- /dev/null +++ b/site/content/docs/use-cases/chat-a2a.md @@ -0,0 +1,147 @@ +--- +title: "A2A Chat" +linkTitle: "A2A Chat" +weight: 30 +--- + +Hold a multi-turn conversation with an [A2A](https://a2a-protocol.org/) +(Agent2Agent) agent hosted on a remote mesh node — using a **stock, +unmodified `a2a-sdk` client**. No SAM-specific client code: the mesh's +agent-card rewrite makes the standard SDK work as-is. + +Source: [`development/examples/chat-a2a/`](https://github.com/google/sam/tree/main/development/examples/chat-a2a). + +## The idea + +A2A is the mesh's southbound agent-to-agent wire: an agent process speaking +A2A over HTTP registers on its node as a `type: a2a` service, and remote +peers reach it through the raw egress path +`/sam/{peer}/a2a/{service}/...` (see +[A2A Service Routing](../../user/node-configuration/#a2a-service-routing)). + +The problem with proxying A2A naively is the **agent card**. A2A clients +bootstrap from `/.well-known/agent-card.json`, and the card contains the +agent's own interface URLs — addresses that are only reachable on the +provider's machine. A stock client would fetch the card through the mesh, +then try to talk to `http://127.0.0.1:7777/` and fail. + +So the caller's node rewrites the card **in transit**: interface URLs are +pointed back at the mesh path the client fetched from, transports the mesh +cannot carry (gRPC needs its own end-to-end connection) are dropped, and +streaming is advertised off. The client follows the rewritten URLs like it +would for any A2A server — discovery, JSON-RPC `message/send`, and +`contextId` round-tripping all work unmodified, while the traffic actually +flows over libp2p between the nodes. + +This example proves both halves: + +- **Card rewrite** — the bundled REPL is a plain `a2a-sdk` client; it works + only because the rewritten card sends it back through the mesh. +- **Conversation continuity** — the agent keeps one Gemini chat session per + A2A `contextId`. Tell it your name, ask for it back two turns later: the + answer shows the context survived the mesh hop. Each turn is still its own + short-lived A2A **task** (`taskId` changes every turn, created and + completed by each `message/send`); `contextId` is what persists and groups + them into one conversation. + +## The pieces + +- **`chat` service** (`agent.py`) — a Gemini-backed A2A agent built on the + Python `a2a-sdk`: serves its agent card, answers `message/send`, and holds + history server-side, one Gemini chat session per `contextId`. The node + proxies to it via `target_url`; the agent itself knows nothing about SAM. +- **REPL client** (`chat.py`) — a ~40-line stock `a2a-sdk` client: resolves + the card through the mesh, then loops `input()` → `message/send`, echoing + the `contextId` the server minted on the first reply. + +## What you can do with it + +- **Bring any A2A agent onto the mesh unchanged** — the example agent is + deliberately vanilla `a2a-sdk`; anything speaking A2A over HTTP (ADK, + LangGraph, a hosted agent) plugs in the same way, one `target_url` line in + the node config. +- **Use any stock A2A client** — `chat.py` is just the smallest one. The + rewritten card means SDKs, CLIs, and other agents resolve and call the + service without knowing SAM exists. +- **Let the agent own the conversation** — the same pattern as + [Gemini Buddy](../gemini-buddy/), but on the standard A2A wire instead of a + custom MCP tool: state lives with the agent, keyed by `contextId`, and + callers only ever send the next message. + +## Try it on kind + +The repository ships a [kind](https://kind.sigs.k8s.io/)-based local mesh +that brings the agent up with one command. + +### 1. Set a Gemini key for the agent image + +`agent.py` calls Gemini, so set your API key on the `ENV GEMINI_API_KEY` +line in `development/examples/chat-a2a/Dockerfile` before building (a free +Google AI Studio key is fine for the demo). + +### 2. Mesh layout + +Host the agent on one node in `development/kind/mesh-config.yaml`: + +```yaml +node-a: # bare node +node-b: chat-a2a # the A2A agent +``` + +### 3. Bring the mesh up and enroll a local caller node + +```bash +make build # builds ./bin/sam-node (once) +make kind-up # control plane + router + agent (node-b) +make kind-local-node # local sam-node enrolled in the mesh — LEAVE RUNNING +``` + +The local node is your entry point: its sidecar API listens on +**`http://127.0.0.1:9099`** (bearer token `devtoken`), MCP tools at `/mcp`. + +### 4. Find the provider peer + +```bash +./bin/mcp-client -url http://127.0.0.1:9099/mcp -token devtoken \ + -tool discover_remote_services -args '{"type":"a2a","name":"chat"}' +``` + +Note the peer ID and export it as `PEER`. Discovery is gossip-fed; retry for +a few seconds after startup if the list comes back empty. + +### 5. Watch the card rewrite happen + +Fetch the agent card through the mesh with nothing but `curl`: + +```bash +curl -s -H 'X-Sam-Authentication: Bearer devtoken' \ + "http://127.0.0.1:9099/sam/$PEER/a2a/chat/.well-known/agent-card.json" | jq +``` + +The interface URLs in the response point back at this +`/sam/{peer}/a2a/chat` path — not at the agent's own `127.0.0.1:7777` — and +`capabilities.streaming` is `false`. That rewrite is the whole trick. + +### 6. Chat + +Requires [`uv`](https://docs.astral.sh/uv/). + +```bash +cd development/examples/chat-a2a +uv run --with-requirements requirements.txt chat.py \ + "http://127.0.0.1:9099/sam/$PEER/a2a/chat" +``` + +Prove the continuity: introduce yourself in one turn, chat about something +else, then ask the agent what your name is. It answers from its own +server-side session — the client never re-sent the history, only the +`contextId`. + +## Configuration + +| var | default | used by | +|-----|---------|---------| +| `GEMINI_API_KEY` | *(placeholder — set in the Dockerfile)* | agent | +| `GEMINI_MODEL` | `models/gemini-3.5-flash-lite` *(set in the Dockerfile)* | agent | + +The listen port (`7777`) is a constant at the top of `agent.py`. diff --git a/site/content/docs/user/node-configuration.md b/site/content/docs/user/node-configuration.md index fb74fd8b..413a6ddb 100644 --- a/site/content/docs/user/node-configuration.md +++ b/site/content/docs/user/node-configuration.md @@ -75,6 +75,7 @@ When configuring `type: a2a` services (Agent2Agent protocol agents): * **URL backends only**: register the agent's local HTTP endpoint as `target_url`. `command` backends are rejected. * **Raw Proxy Access**: remote peers reach the agent at `http://localhost:8080/sam/{peer}/a2a/{service}/...`. The agent card served at `.../.well-known/agent-card.json` is rewritten in transit so its interface URLs point back at this mesh path; transports the mesh cannot carry (gRPC) are dropped and streaming is advertised off. * **Label-gated egress**: setting `X-Sam-Required-Labels: key=value[,key=value]` on a raw a2a request makes the caller's node verify the provider's control-plane-attested labels and refuse fail-closed (HTTP 403) before any data leaves the node. The header is stripped before forwarding. +* **Runnable example**: the [A2A Chat use case](../../use-cases/chat-a2a/) walks through hosting an a2a agent on a kind mesh and talking to it with a stock `a2a-sdk` client. --- From 9bebe5975a24980db8253199411813c89187632d Mon Sep 17 00:00:00 2001 From: Tomas Tormo Date: Tue, 1 Sep 2026 20:52:05 +0000 Subject: [PATCH 14/16] tests: connect peers via the debug endpoint helper in the a2a CUJ --- tests/integration/a2a_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/a2a_test.go b/tests/integration/a2a_test.go index 5779b0fb..1d317223 100644 --- a/tests/integration/a2a_test.go +++ b/tests/integration/a2a_test.go @@ -63,7 +63,7 @@ func TestA2ACUJ(t *testing.T) { waitForAPI(t, apiAddrB) addrA := waitForPeerInfoInLog(t, filepath.Join(homeA, "node.log")) - callMCP(t, apiAddrB, "connect_peer", map[string]any{"peer_addr": addrA}) + connectPeer(t, apiAddrB, addrA) waitForDHTPeers(t, apiAddrA) idx := strings.LastIndex(addrA, "/p2p/") From fb5c34999391521c8b02e4170a53a227986d3995 Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Wed, 2 Sep 2026 08:43:31 +0000 Subject: [PATCH 15/16] node: regenerate a2a agent cards with the official SDK instead of live rewrite The caller's node now impersonates the remote agent's card endpoint: it holds the client request, fetches the card from the agent over the mesh, and serves a card regenerated from typed a2aproject/a2a-go/v2 structs whose interfaces point at the local mesh path. This replaces the ReverseProxy ModifyResponse in-flight rewrite, the context-key plumbing that carried the mesh base URL, and the content-encoding bail-out (the node negotiates identity encoding on its own fetch). Typed regeneration also drops the card's original JWS signatures, which no longer verify once the content changes, and fails closed with an explicit 502 on cards that advertise no binding the mesh can carry (gRPC-only or pre-1.0 cards). --- development/examples/chat-a2a/agent.py | 2 +- go.mod | 1 + go.sum | 2 + internal/node/a2a_service.go | 165 ++++++++----- internal/node/a2a_service_test.go | 247 +++++++++++-------- internal/node/sidecar.go | 80 ++++-- site/content/docs/use-cases/chat-a2a.md | 28 ++- site/content/docs/user/node-configuration.md | 2 +- tests/integration/a2a_test.go | 25 +- 9 files changed, 333 insertions(+), 219 deletions(-) diff --git a/development/examples/chat-a2a/agent.py b/development/examples/chat-a2a/agent.py index ec4ec8a6..b3802826 100644 --- a/development/examples/chat-a2a/agent.py +++ b/development/examples/chat-a2a/agent.py @@ -96,7 +96,7 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None agent_card=agent_card, ) # Starlette over FastAPI: the SDK generates the routes, so FastAPI would add nothing. -# JSON-RPC at "/": the mesh card rewrite drops URL subpaths, so clients land on the root. +# JSON-RPC at "/": the mesh card regeneration drops URL subpaths, so clients land on the root. app = Starlette( routes=[ *create_jsonrpc_routes(request_handler=handler, rpc_url="/"), diff --git a/go.mod b/go.mod index ef6900e6..9d91fe41 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/google/sam go 1.25.7 require ( + github.com/a2aproject/a2a-go/v2 v2.5.0 github.com/biscuit-auth/biscuit-go/v2 v2.2.0 github.com/coreos/go-oidc/v3 v3.20.0 github.com/golang-jwt/jwt/v5 v5.3.1 diff --git a/go.sum b/go.sum index 2a449f98..5e185644 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,8 @@ filippo.io/bigmod v0.1.1-0.20260103110540-f8a47775ebe5 h1:JA0fFr+kxpqTdxR9LOBiTW filippo.io/bigmod v0.1.1-0.20260103110540-f8a47775ebe5/go.mod h1:OjOXDNlClLblvXdwgFFOQFJEocLhhtai8vGLy0JCZlI= filippo.io/keygen v1.0.0 h1:u0/Fhxlgz3uPv+XxhfgTq3BJt5VesIPM5ue/OuG7qjQ= filippo.io/keygen v1.0.0/go.mod h1:9nnw1SlYHYuPSo/3wjQzNjSbeHlq2NsKo5iEtfJPWP0= +github.com/a2aproject/a2a-go/v2 v2.5.0 h1:ZdcFoxv+nZTUV0i2ue5hES76YCANFPG9vjqd7vK8yWM= +github.com/a2aproject/a2a-go/v2 v2.5.0/go.mod h1:NcRp/ZHxgMzDj12/BteIC2gOjljuEBKaGRfEdJ2lNSI= github.com/alecthomas/assert/v2 v2.11.0 h1:2Q9r3ki8+JYXvGsDyBXwH3LcJ+WK5D0gc5E8vS6K3D0= github.com/alecthomas/assert/v2 v2.11.0/go.mod h1:Bze95FyfUr7x34QZrjL+XP+0qgp/zg8yS+TtBj1WA3k= github.com/alecthomas/participle/v2 v2.1.4 h1:W/H79S8Sat/krZ3el6sQMvMaahJ+XcM9WSI2naI7w2U= diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index f3282907..22a9a808 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -15,7 +15,6 @@ package node import ( - "bytes" "context" "encoding/json" "fmt" @@ -24,14 +23,15 @@ import ( "strconv" "strings" + "github.com/a2aproject/a2a-go/v2/a2a" "github.com/google/sam/api" "github.com/libp2p/go-libp2p/core/peer" ) func init() { registerEgressMiddleware(api.ServiceTypeStringA2A, egressMiddleware{ - gateRequest: a2aEgressGate, - modifyResponse: rewriteA2AAgentCard, + gateRequest: a2aEgressGate, + serveLocal: a2aServeAgentCard, }) } @@ -56,13 +56,9 @@ func (s *A2AService) Init(ctx context.Context) error { return nil } -// a2aCardBaseURL is the context key carrying the caller-facing mesh base URL -// of an agent-card fetch, set by a2aEgressGate and consumed by the rewrite. -type a2aCardBaseURL struct{} - // a2aEgressGate runs the caller-side A2A checks on a raw egress request: -// the fail-closed labels gate and tagging agent-card fetches for rewrite. -// On refusal it writes the HTTP error itself and returns ok=false. +// the fail-closed labels gate. On refusal it writes the HTTP error itself +// and returns ok=false. func a2aEgressGate(node *SamNode, w http.ResponseWriter, r *http.Request, route egressRoute) (*http.Request, bool) { if labelsHeader := r.Header.Get(api.HeaderSamRequiredLabels); labelsHeader != "" { r.Header.Del(api.HeaderSamRequiredLabels) @@ -82,78 +78,115 @@ func a2aEgressGate(node *SamNode, w http.ResponseWriter, r *http.Request, route return r, false } } - if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/.well-known/agent-card.json") { - base := fmt.Sprintf("http://%s/sam/%s/%s/%s", r.Host, route.peerID, route.serviceType, route.serviceName) - r = r.WithContext(context.WithValue(r.Context(), a2aCardBaseURL{}, base)) - } return r, true } -// rewriteA2AAgentCard makes a proxied agent card usable by stock A2A clients: -// interface URLs point back at the mesh path, transports the mesh cannot -// carry (gRPC) are dropped, and streaming is advertised off until verified. -func rewriteA2AAgentCard(resp *http.Response) error { - base, ok := resp.Request.Context().Value(a2aCardBaseURL{}).(string) - if !ok || resp.StatusCode != http.StatusOK { - return nil - } - if resp.Header.Get("Content-Encoding") != "" { - logger.Warnf("[A2A] agent card response is content-encoded; skipping rewrite") - return nil +// a2aAgentCardPath is the well-known agent card location (A2A spec / RFC 8615). +const a2aAgentCardPath = ".well-known/agent-card.json" + +// maxAgentCardBytes bounds how much of a remote agent card the node ingests. +const maxAgentCardBytes = 1 << 20 + +// a2aServeAgentCard impersonates the remote agent's card endpoint: it holds +// the client request, fetches the card from the agent over the mesh, and +// serves a regenerated card whose interfaces point at the local mesh URL. +// Stock A2A clients then talk to the agent through this node unmodified. +// Non-card requests are left to the streaming egress proxy. +func a2aServeAgentCard(node *SamNode, rt http.RoundTripper, w http.ResponseWriter, r *http.Request, route egressRoute) bool { + if r.Method != http.MethodGet || route.upstreamPath != a2aAgentCardPath { + return false } - body, err := io.ReadAll(resp.Body) - _ = resp.Body.Close() + resp, err := fetchRemoteAgentCard(node, rt, r, route) if err != nil { - return err + logger.Warnf("[A2A] agent card fetch from %s failed: %v", route.peerID, err) + http.Error(w, "Bad Gateway: agent card fetch failed", http.StatusBadGateway) + return true } - var card map[string]any - if err := json.Unmarshal(body, &card); err != nil { - return fmt.Errorf("agent card is not valid JSON: %w", err) + defer func() { _ = resp.Body.Close() }() + + body := io.LimitReader(resp.Body, maxAgentCardBytes) + if resp.StatusCode != http.StatusOK { + // The agent's own error; relay it as-is. + if ct := resp.Header.Get("Content-Type"); ct != "" { + w.Header().Set("Content-Type", ct) + } + w.WriteHeader(resp.StatusCode) + _, _ = io.Copy(w, body) + return true } - if _, ok := card["url"]; ok { - card["url"] = base + + var card a2a.AgentCard + if err := json.NewDecoder(body).Decode(&card); err != nil { + logger.Warnf("[A2A] agent card from %s is not valid JSON: %v", route.peerID, err) + http.Error(w, "Bad Gateway: agent card is not valid JSON", http.StatusBadGateway) + return true } - if pt, ok := card["preferredTransport"].(string); ok && !a2aTransportOverHTTP(pt) { - card["preferredTransport"] = "JSONRPC" + base := fmt.Sprintf("http://%s/sam/%s/%s/%s", r.Host, route.peerID, route.serviceType, route.serviceName) + if err := regenerateAgentCardForMesh(&card, base); err != nil { + logger.Warnf("[A2A] agent card from %s unusable through the mesh: %v", route.peerID, err) + http.Error(w, fmt.Sprintf("Bad Gateway: %v", err), http.StatusBadGateway) + return true } - for _, key := range []string{"additionalInterfaces", "supportedInterfaces"} { - ifaces, ok := card[key].([]any) - if !ok { - continue - } - kept := make([]any, 0, len(ifaces)) - for _, entry := range ifaces { - iface, ok := entry.(map[string]any) - if !ok { - continue - } - transport, _ := iface["transport"].(string) - if transport == "" { - transport, _ = iface["protocolBinding"].(string) - } - if !a2aTransportOverHTTP(transport) { - continue - } - iface["url"] = base - kept = append(kept, iface) - } - card[key] = kept + out, err := json.Marshal(&card) + if err != nil { + http.Error(w, "Internal Server Error", http.StatusInternalServerError) + return true } - if caps, ok := card["capabilities"].(map[string]any); ok { - caps["streaming"] = false + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Length", strconv.Itoa(len(out))) + _, _ = w.Write(out) + return true +} + +// fetchRemoteAgentCard performs the mesh-side GET for the agent card, reusing +// the headers already prepared for egress (biscuit, agent claim, passthrough +// Authorization) on the incoming request. +func fetchRemoteAgentCard(node *SamNode, rt http.RoundTripper, r *http.Request, route egressRoute) (*http.Response, error) { + ctx := allowLimitedEgressConn(r.Context()) + if node != nil { + node.prepareEgressPeer(ctx, route.peerID) } - out, err := json.Marshal(card) + url := fmt.Sprintf("libp2p://%s/%s/%s/%s", route.peerID, route.serviceType, route.serviceName, a2aAgentCardPath) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { - return err + return nil, err + } + req.Header = r.Header.Clone() + // The node decodes the card itself, so negotiate identity encoding + // regardless of what the held client asked for. + req.Header.Del("Accept-Encoding") + req.Host = route.peerID + return (&http.Client{Transport: rt}).Do(req) +} + +// regenerateAgentCardForMesh rebuilds a fetched agent card for mesh use: +// interface URLs point back at the mesh path, bindings the mesh cannot carry +// (gRPC) are dropped, streaming is advertised off until verified, and the +// original signatures are removed since they no longer match the content. +func regenerateAgentCardForMesh(card *a2a.AgentCard, base string) error { + kept := make([]*a2a.AgentInterface, 0, len(card.SupportedInterfaces)) + for _, iface := range card.SupportedInterfaces { + if iface == nil || !a2aBindingOverHTTP(iface.ProtocolBinding) { + continue + } + iface.URL = base + kept = append(kept, iface) } - resp.Body = io.NopCloser(bytes.NewReader(out)) - resp.ContentLength = int64(len(out)) - resp.Header.Set("Content-Length", strconv.Itoa(len(out))) + if len(kept) == 0 { + return fmt.Errorf("agent card advertises no supported interface the mesh can carry (JSONRPC or HTTP+JSON); is the agent serving a pre-1.0 A2A card?") + } + card.SupportedInterfaces = kept + card.Capabilities.Streaming = false + card.Signatures = nil return nil } -// a2aTransportOverHTTP reports whether an A2A transport can traverse the +// a2aBindingOverHTTP reports whether an A2A protocol binding can traverse the // mesh's HTTP-over-libp2p path; gRPC needs its own end-to-end connection. -func a2aTransportOverHTTP(transport string) bool { - return transport == "JSONRPC" || transport == "HTTP+JSON" +func a2aBindingOverHTTP(binding a2a.TransportProtocol) bool { + switch strings.ToUpper(string(binding)) { + case string(a2a.TransportProtocolJSONRPC), string(a2a.TransportProtocolHTTPJSON): + return true + } + return false } diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index 9a4c38f5..d8f80126 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -17,12 +17,14 @@ package node import ( "context" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" "strings" "testing" + "github.com/a2aproject/a2a-go/v2/a2a" "github.com/google/sam/api" ) @@ -89,87 +91,103 @@ func TestA2AEgressHookMalformedLabels(t *testing.T) { } } -func TestA2AEgressHookTagsCardFetch(t *testing.T) { - rec := httptest.NewRecorder() - req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) - req.Host = "127.0.0.1:8080" - r2, ok := applyEgressMiddleware(nil, rec, req) - if !ok { - t.Fatal("card fetch must pass through") - } - base, _ := r2.Context().Value(a2aCardBaseURL{}).(string) - want := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" - if base != want { - t.Fatalf("card base = %q, want %q", base, want) +// roundTripFunc fakes the mesh transport for agent-card fetches. +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func cardResponse(status int, body string) *http.Response { + return &http.Response{ + StatusCode: status, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(body)), } } -func TestRewriteA2AAgentCard(t *testing.T) { - card := `{ +func TestA2AServeAgentCardRegenerates(t *testing.T) { + const upstreamCard = `{ "name": "T", - "url": "http://localhost:9999", - "preferredTransport": "GRPC", - "additionalInterfaces": [ - {"url": "http://localhost:9999", "transport": "JSONRPC"}, - {"url": "localhost:50051", "transport": "GRPC"} - ], + "description": "d", + "version": "1.0.0", + "capabilities": {"streaming": true}, "supportedInterfaces": [ - {"url": "http://localhost:9999", "protocolBinding": "JSONRPC"}, - {"url": "localhost:50051", "protocolBinding": "GRPC"} + {"url": "http://localhost:7777/", "protocolBinding": "JSONRPC", "protocolVersion": "1.0"}, + {"url": "localhost:50051", "protocolBinding": "GRPC", "protocolVersion": "1.0"} ], - "capabilities": {"streaming": true} + "signatures": [{"protected": "eyJh", "signature": "sig"}], + "defaultInputModes": ["text"], + "defaultOutputModes": ["text"], + "skills": [] }` - base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + var outbound *http.Request + rt := roundTripFunc(func(r *http.Request) (*http.Response, error) { + outbound = r + return cardResponse(http.StatusOK, upstreamCard), nil + }) + + rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) - req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) - resp := &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{}, - Body: io.NopCloser(strings.NewReader(card)), - Request: req, - } - if err := rewriteA2AAgentCard(resp); err != nil { - t.Fatal(err) - } - body, _ := io.ReadAll(resp.Body) - var got map[string]any - if err := json.Unmarshal(body, &got); err != nil { - t.Fatalf("rewritten card is not JSON: %v", err) - } - if got["url"] != base { - t.Errorf("url = %v, want %s", got["url"], base) - } - if got["preferredTransport"] != "JSONRPC" { - t.Errorf("preferredTransport = %v, want JSONRPC", got["preferredTransport"]) - } - for _, key := range []string{"additionalInterfaces", "supportedInterfaces"} { - ifaces, _ := got[key].([]any) - if len(ifaces) != 1 { - t.Fatalf("%s: want 1 HTTP interface after dropping gRPC, got %v", key, got[key]) - } - if u := ifaces[0].(map[string]any)["url"]; u != base { - t.Errorf("%s url = %v, want %s", key, u, base) - } + req.Host = "127.0.0.1:8080" + req.Header.Set(api.HeaderSamBiscuit, "b64-biscuit") + req.Header.Set("Accept-Encoding", "gzip") + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("agent-card GET must be handled locally") + } + + wantURL := "libp2p://12D3KooWpeer/a2a/agent/.well-known/agent-card.json" + if got := outbound.URL.String(); got != wantURL { + t.Errorf("outbound fetch URL = %q, want %q", got, wantURL) } - if s := got["capabilities"].(map[string]any)["streaming"]; s != false { - t.Errorf("streaming = %v, want false", s) + if outbound.Header.Get(api.HeaderSamBiscuit) != "b64-biscuit" { + t.Error("egress headers must be carried on the card fetch") + } + if outbound.Header.Get("Accept-Encoding") != "" { + t.Error("card fetch must negotiate identity encoding") } -} -func TestRewriteA2AAgentCardNoopWithoutTag(t *testing.T) { - orig := `{"name":"T","url":"http://localhost:9999"}` - resp := &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{}, - Body: io.NopCloser(strings.NewReader(orig)), - Request: httptest.NewRequest("GET", "/anything", nil), + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body: %s", rec.Code, rec.Body.String()) + } + var card a2a.AgentCard + if err := json.Unmarshal(rec.Body.Bytes(), &card); err != nil { + t.Fatalf("regenerated card is not a valid AgentCard: %v", err) + } + base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" + if len(card.SupportedInterfaces) != 1 { + t.Fatalf("want 1 HTTP interface after dropping gRPC, got %v", card.SupportedInterfaces) + } + if card.SupportedInterfaces[0].URL != base { + t.Errorf("interface url = %q, want %q", card.SupportedInterfaces[0].URL, base) + } + if card.SupportedInterfaces[0].ProtocolBinding != a2a.TransportProtocolJSONRPC { + t.Errorf("binding = %q, want JSONRPC", card.SupportedInterfaces[0].ProtocolBinding) + } + if card.Capabilities.Streaming { + t.Error("streaming must be advertised off through the mesh") } - if err := rewriteA2AAgentCard(resp); err != nil { - t.Fatal(err) + if len(card.Signatures) != 0 { + t.Error("stale signatures must be dropped from the regenerated card") } - body, _ := io.ReadAll(resp.Body) - if string(body) != orig { - t.Fatalf("untagged response was modified: %s", body) + if card.Name != "T" || card.Version != "1.0.0" { + t.Errorf("agent identity fields must survive regeneration: %+v", card) + } +} + +func TestA2AServeAgentCardIgnoresNonCardRequests(t *testing.T) { + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + t.Fatal("no mesh fetch expected") + return nil, nil + }) + for _, tc := range []struct{ method, path string }{ + {"POST", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json"}, + {"GET", "/sam/12D3KooWpeer/a2a/agent/tasks/1"}, + {"GET", "/sam/12D3KooWpeer/mcp/svc/.well-known/agent-card.json"}, + } { + rec := httptest.NewRecorder() + req := httptest.NewRequest(tc.method, tc.path, nil) + if serveEgressLocally(nil, rt, rec, req) { + t.Errorf("%s %s must stream through the proxy", tc.method, tc.path) + } } } @@ -186,43 +204,76 @@ func TestA2AEgressHookInvalidPeerID(t *testing.T) { } } -func TestRewriteA2AAgentCardSkipsNon200(t *testing.T) { - orig := `{"name":"T","url":"http://localhost:9999"}` - base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" +func TestA2AServeAgentCardNoHTTPBindingFailsClosed(t *testing.T) { + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + return cardResponse(http.StatusOK, + `{"name":"T","supportedInterfaces":[{"url":"localhost:50051","protocolBinding":"GRPC"}]}`), nil + }) + rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) - req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) - resp := &http.Response{ - StatusCode: http.StatusNotFound, - Header: http.Header{}, - Body: io.NopCloser(strings.NewReader(orig)), - Request: req, + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("card GET must be handled locally") } - if err := rewriteA2AAgentCard(resp); err != nil { - t.Fatal(err) + if rec.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", rec.Code) } - body, _ := io.ReadAll(resp.Body) - if string(body) != orig { - t.Fatalf("non-200 response was modified: %s", body) + if !strings.Contains(rec.Body.String(), "mesh can carry") { + t.Errorf("error must name the refusal reason, got: %s", rec.Body.String()) } } -func TestRewriteA2AAgentCardSkipsContentEncoded(t *testing.T) { - orig := `{"name":"T","url":"http://localhost:9999"}` - base := "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" +func TestA2AServeAgentCardPre10CardFailsClosed(t *testing.T) { + // A2A v0.3-shaped card: interfaces live in additionalInterfaces, which the + // v1.0 type does not carry, so regeneration must refuse rather than serve + // a card with no usable interface. + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + return cardResponse(http.StatusOK, `{"name":"T","url":"http://localhost:9999",`+ + `"preferredTransport":"JSONRPC",`+ + `"additionalInterfaces":[{"url":"http://localhost:9999","transport":"JSONRPC"}]}`), nil + }) + rec := httptest.NewRecorder() req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) - req = req.WithContext(context.WithValue(req.Context(), a2aCardBaseURL{}, base)) - resp := &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{"Content-Encoding": []string{"gzip"}}, - Body: io.NopCloser(strings.NewReader(orig)), - Request: req, - } - if err := rewriteA2AAgentCard(resp); err != nil { - t.Fatal(err) - } - body, _ := io.ReadAll(resp.Body) - if string(body) != orig { - t.Fatalf("content-encoded response was modified: %s", body) + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("card GET must be handled locally") + } + if rec.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", rec.Code) + } + if !strings.Contains(rec.Body.String(), "pre-1.0") { + t.Errorf("error must hint at the card vintage, got: %s", rec.Body.String()) + } +} + +func TestA2AServeAgentCardRelaysUpstreamError(t *testing.T) { + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + resp := cardResponse(http.StatusNotFound, "no card here") + resp.Header.Set("Content-Type", "text/plain") + return resp, nil + }) + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("card GET must be handled locally") + } + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want the agent's own 404", rec.Code) + } + if !strings.Contains(rec.Body.String(), "no card here") { + t.Errorf("agent's error body must be relayed, got: %s", rec.Body.String()) + } +} + +func TestA2AServeAgentCardFetchErrorIs502(t *testing.T) { + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("peer unreachable") + }) + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("card GET must be handled locally") + } + if rec.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", rec.Code) } } diff --git a/internal/node/sidecar.go b/internal/node/sidecar.go index 4b6992ab..b88d5bf4 100644 --- a/internal/node/sidecar.go +++ b/internal/node/sidecar.go @@ -659,8 +659,10 @@ type egressMiddleware struct { // gateRequest may refuse the request (writing the HTTP error itself, // returning false) or return a derived request to forward. gateRequest func(node *SamNode, w http.ResponseWriter, r *http.Request, route egressRoute) (*http.Request, bool) - // modifyResponse edits the proxied response; nil means no hook. - modifyResponse func(*http.Response) error + // serveLocal may fully handle the request at this node (returning true) + // instead of letting it stream through the proxy; nil means no hook. + // It runs after the egress headers (biscuit, agent claim) are prepared. + serveLocal func(node *SamNode, rt http.RoundTripper, w http.ResponseWriter, r *http.Request, route egressRoute) bool } // egressRoute is the parsed /sam/{peer}/{type}/{svc}/{upstream} egress path, @@ -679,17 +681,26 @@ func registerEgressMiddleware(serviceType string, mw egressMiddleware) { egressMiddlewares[strings.ToLower(serviceType)] = mw } -// applyEgressMiddleware routes a raw egress request through the middleware -// registered for its service type, if any. -func applyEgressMiddleware(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { - parts := strings.SplitN(r.URL.Path, "/", 6) +// parseEgressRoute parses a /sam/{peer}/{type}/{svc}/{upstream} egress path. +func parseEgressRoute(path string) (egressRoute, bool) { + parts := strings.SplitN(path, "/", 6) if len(parts) < 5 { - return r, true + return egressRoute{}, false } route := egressRoute{peerID: parts[2], serviceType: strings.ToLower(parts[3]), serviceName: parts[4]} if len(parts) > 5 { route.upstreamPath = parts[5] } + return route, true +} + +// applyEgressMiddleware routes a raw egress request through the middleware +// registered for its service type, if any. +func applyEgressMiddleware(node *SamNode, w http.ResponseWriter, r *http.Request) (*http.Request, bool) { + route, ok := parseEgressRoute(r.URL.Path) + if !ok { + return r, true + } mw, ok := egressMiddlewares[route.serviceType] if !ok || mw.gateRequest == nil { return r, true @@ -697,15 +708,37 @@ func applyEgressMiddleware(node *SamNode, w http.ResponseWriter, r *http.Request return mw.gateRequest(node, w, r, route) } -// egressModifyResponse dispatches to the middleware of the service type in -// the rewritten path (/{type}/{svc}/...); types without a hook pass through. -func egressModifyResponse(resp *http.Response) error { - parts := strings.SplitN(strings.TrimPrefix(resp.Request.URL.Path, "/"), "/", 2) - mw, ok := egressMiddlewares[strings.ToLower(parts[0])] - if !ok || mw.modifyResponse == nil { - return nil +// serveEgressLocally gives the service type's middleware a chance to answer +// the request from this node (e.g. agent-card regeneration) instead of +// proxying; it reports whether the request was handled. +func serveEgressLocally(node *SamNode, rt http.RoundTripper, w http.ResponseWriter, r *http.Request) bool { + route, ok := parseEgressRoute(r.URL.Path) + if !ok { + return false + } + mw, ok := egressMiddlewares[route.serviceType] + if !ok || mw.serveLocal == nil { + return false + } + return mw.serveLocal(node, rt, w, r, route) +} + +// allowLimitedEgressConn lets egress traffic ride limited (relayed) libp2p +// connections while a direct one is established. +func allowLimitedEgressConn(ctx context.Context) context.Context { + return network.WithAllowLimitedConn(ctx, "egress-proxy") +} + +// prepareEgressPeer warms up connectivity to the destination peer of an +// egress request if it is not already connected. +func (node *SamNode) prepareEgressPeer(ctx context.Context, peerID string) { + pid, err := peer.Decode(peerID) + if err != nil { + return + } + if cond := node.Host.Network().Connectedness(pid); cond != network.Connected && cond != network.Limited { + node.preparePeerAddrs(ctx, pid) } - return mw.modifyResponse(resp) } func createEgressProxy(node *SamNode) http.Handler { @@ -713,8 +746,7 @@ func createEgressProxy(node *SamNode) http.Handler { proxy := &httputil.ReverseProxy{ Director: func(req *http.Request) { - ctx := req.Context() - ctx = network.WithAllowLimitedConn(ctx, "egress-proxy") + ctx := allowLimitedEgressConn(req.Context()) *req = *req.WithContext(ctx) parts := strings.SplitN(req.URL.Path, "/", 6) @@ -722,12 +754,7 @@ func createEgressProxy(node *SamNode) http.Handler { return } peerID := parts[2] - pid, err := peer.Decode(peerID) - if err == nil { - if cond := node.Host.Network().Connectedness(pid); cond != network.Connected && cond != network.Limited { - node.preparePeerAddrs(ctx, pid) - } - } + node.prepareEgressPeer(ctx, peerID) serviceType := parts[3] serviceName := parts[4] upstreamPath := "" @@ -747,8 +774,7 @@ func createEgressProxy(node *SamNode) http.Handler { req.URL.RawPath = "" logger.Debugf("[Proxy] Rewriting URL to libp2p://%s%s", req.URL.Host, req.URL.Path) }, - Transport: transport, - ModifyResponse: egressModifyResponse, + Transport: transport, } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -785,6 +811,10 @@ func createEgressProxy(node *SamNode) http.Handler { // "Authorization" header passes straight through untouched as the destination's own credential. r.Header.Del(api.HeaderSamAuthentication) + if serveEgressLocally(node, transport, w, r) { + return + } + proxy.ServeHTTP(w, r) }) } diff --git a/site/content/docs/use-cases/chat-a2a.md b/site/content/docs/use-cases/chat-a2a.md index accb519e..028d8590 100644 --- a/site/content/docs/use-cases/chat-a2a.md +++ b/site/content/docs/use-cases/chat-a2a.md @@ -7,7 +7,7 @@ weight: 30 Hold a multi-turn conversation with an [A2A](https://a2a-protocol.org/) (Agent2Agent) agent hosted on a remote mesh node — using a **stock, unmodified `a2a-sdk` client**. No SAM-specific client code: the mesh's -agent-card rewrite makes the standard SDK work as-is. +agent-card regeneration makes the standard SDK work as-is. Source: [`development/examples/chat-a2a/`](https://github.com/google/sam/tree/main/development/examples/chat-a2a). @@ -25,18 +25,19 @@ agent's own interface URLs — addresses that are only reachable on the provider's machine. A stock client would fetch the card through the mesh, then try to talk to `http://127.0.0.1:7777/` and fail. -So the caller's node rewrites the card **in transit**: interface URLs are -pointed back at the mesh path the client fetched from, transports the mesh -cannot carry (gRPC needs its own end-to-end connection) are dropped, and -streaming is advertised off. The client follows the rewritten URLs like it -would for any A2A server — discovery, JSON-RPC `message/send`, and -`contextId` round-tripping all work unmodified, while the traffic actually -flows over libp2p between the nodes. +So the caller's node **impersonates the card endpoint**: it holds the +client's request, fetches the card from the agent over the mesh, and serves +a regenerated card — interface URLs point back at the mesh path the client +fetched from, protocol bindings the mesh cannot carry (gRPC needs its own +end-to-end connection) are dropped, and streaming is advertised off. The +client follows the regenerated card like it would for any A2A server — +discovery, JSON-RPC `message/send`, and `contextId` round-tripping all work +unmodified, while the traffic actually flows over libp2p between the nodes. This example proves both halves: -- **Card rewrite** — the bundled REPL is a plain `a2a-sdk` client; it works - only because the rewritten card sends it back through the mesh. +- **Card regeneration** — the bundled REPL is a plain `a2a-sdk` client; it + works only because the regenerated card sends it back through the mesh. - **Conversation continuity** — the agent keeps one Gemini chat session per A2A `contextId`. Tell it your name, ask for it back two turns later: the answer shows the context survived the mesh hop. Each turn is still its own @@ -61,7 +62,7 @@ This example proves both halves: LangGraph, a hosted agent) plugs in the same way, one `target_url` line in the node config. - **Use any stock A2A client** — `chat.py` is just the smallest one. The - rewritten card means SDKs, CLIs, and other agents resolve and call the + regenerated card means SDKs, CLIs, and other agents resolve and call the service without knowing SAM exists. - **Let the agent own the conversation** — the same pattern as [Gemini Buddy](../gemini-buddy/), but on the standard A2A wire instead of a @@ -109,7 +110,7 @@ The local node is your entry point: its sidecar API listens on Note the peer ID and export it as `PEER`. Discovery is gossip-fed; retry for a few seconds after startup if the list comes back empty. -### 5. Watch the card rewrite happen +### 5. Watch the card regeneration happen Fetch the agent card through the mesh with nothing but `curl`: @@ -120,7 +121,8 @@ curl -s -H 'X-Sam-Authentication: Bearer devtoken' \ The interface URLs in the response point back at this `/sam/{peer}/a2a/chat` path — not at the agent's own `127.0.0.1:7777` — and -`capabilities.streaming` is `false`. That rewrite is the whole trick. +`capabilities.streaming` is `false`. That regenerated card is the whole +trick. ### 6. Chat diff --git a/site/content/docs/user/node-configuration.md b/site/content/docs/user/node-configuration.md index 413a6ddb..1c57c1b5 100644 --- a/site/content/docs/user/node-configuration.md +++ b/site/content/docs/user/node-configuration.md @@ -73,7 +73,7 @@ When configuring `target_url` for `type: inference` services (e.g. Ollama, vLLM, When configuring `type: a2a` services (Agent2Agent protocol agents): * **URL backends only**: register the agent's local HTTP endpoint as `target_url`. `command` backends are rejected. -* **Raw Proxy Access**: remote peers reach the agent at `http://localhost:8080/sam/{peer}/a2a/{service}/...`. The agent card served at `.../.well-known/agent-card.json` is rewritten in transit so its interface URLs point back at this mesh path; transports the mesh cannot carry (gRPC) are dropped and streaming is advertised off. +* **Raw Proxy Access**: remote peers reach the agent at `http://localhost:8080/sam/{peer}/a2a/{service}/...`. For the agent card at `.../.well-known/agent-card.json`, the caller's node impersonates the endpoint: it fetches the card from the agent (A2A v1.0 format) and serves a regenerated one whose interface URLs point back at this mesh path; protocol bindings the mesh cannot carry (gRPC) are dropped, streaming is advertised off, and the original signatures are removed since the content changed. * **Label-gated egress**: setting `X-Sam-Required-Labels: key=value[,key=value]` on a raw a2a request makes the caller's node verify the provider's control-plane-attested labels and refuse fail-closed (HTTP 403) before any data leaves the node. The header is stripped before forwarding. * **Runnable example**: the [A2A Chat use case](../../use-cases/chat-a2a/) walks through hosting an a2a agent on a kind mesh and talking to it with a stock `a2a-sdk` client. diff --git a/tests/integration/a2a_test.go b/tests/integration/a2a_test.go index 1d317223..721ea035 100644 --- a/tests/integration/a2a_test.go +++ b/tests/integration/a2a_test.go @@ -82,10 +82,10 @@ func TestA2ACUJ(t *testing.T) { switch { case r.Method == http.MethodGet && r.URL.Path == "/.well-known/agent-card.json": w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"name":"echo-agent","url":"http://localhost:9999",` + - `"preferredTransport":"JSONRPC",` + - `"additionalInterfaces":[{"url":"http://localhost:9999","transport":"JSONRPC"},` + - `{"url":"localhost:50051","transport":"GRPC"}],` + + _, _ = w.Write([]byte(`{"name":"echo-agent",` + + `"supportedInterfaces":[` + + `{"url":"http://localhost:9999","protocolBinding":"JSONRPC","protocolVersion":"1.0"},` + + `{"url":"localhost:50051","protocolBinding":"GRPC","protocolVersion":"1.0"}],` + `"capabilities":{"streaming":true}}`)) case r.Method == http.MethodPost: sendCount.Add(1) @@ -123,12 +123,10 @@ func TestA2ACUJ(t *testing.T) { time.Sleep(200 * time.Millisecond) } var card struct { - URL string `json:"url"` - PreferredTransport string `json:"preferredTransport"` - AdditionalInterfaces []struct { - URL string `json:"url"` - Transport string `json:"transport"` - } `json:"additionalInterfaces"` + SupportedInterfaces []struct { + URL string `json:"url"` + ProtocolBinding string `json:"protocolBinding"` + } `json:"supportedInterfaces"` Capabilities struct { Streaming bool `json:"streaming"` } `json:"capabilities"` @@ -136,11 +134,8 @@ func TestA2ACUJ(t *testing.T) { if err := json.Unmarshal(cardBody, &card); err != nil { t.Fatalf("invalid card: %v, body: %s", err, string(cardBody)) } - if card.URL != meshBase { - t.Errorf("card url = %q, want mesh base %q", card.URL, meshBase) - } - if len(card.AdditionalInterfaces) != 1 || card.AdditionalInterfaces[0].URL != meshBase { - t.Errorf("interfaces not rewritten / gRPC not dropped: %s", string(cardBody)) + if len(card.SupportedInterfaces) != 1 || card.SupportedInterfaces[0].URL != meshBase { + t.Errorf("interfaces not regenerated / gRPC not dropped: %s", string(cardBody)) } if card.Capabilities.Streaming { t.Error("streaming must be advertised off through the mesh") From a3d9c4f60e23a0951f17c609a1a9d263e3f351df Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Wed, 2 Sep 2026 09:52:34 +0000 Subject: [PATCH 16/16] a2a: cover the CUJ with stock SDKs at every level and fix what that found The integration CUJ now runs the official SDK on both ends (a2asrv echo agent, a2aclient + card resolver through the mesh) and a new e2e bats CUJ drives a containerized mesh with the stock python a2a-sdk on both ends, bootstrapping from the regenerated card and exercising the labels gate over the real control-plane attestation chain. Testing with real SDK parsers instead of hand-rolled JSON surfaced three fixes: - regenerated cards kept nil required list fields, which encoding/json marshals as null and strict card parsers (pydantic) reject; they are now normalized to empty arrays. - the official Go resolver treats a pathful base URL as the card URL itself (the python resolver appends the well-known path), so the regenerated card is now served at the bare service root as well; a root GET is not part of any A2A binding, JSON-RPC being POST-only. - the Helm bootstrap job had no way to grant allowed_labels, so no chart-deployed mesh could enroll a labelled node at all; a new bootstrap.nodeLabels value carries the grant patterns, empty (fail closed) by default. --- charts/sam-mesh/templates/bootstrap-job.yaml | 2 +- charts/sam-mesh/tests/bootstrap-job_test.yaml | 21 +++ charts/sam-mesh/values.yaml | 3 + internal/node/a2a_service.go | 19 +- internal/node/a2a_service_test.go | 62 +++++++ tests/e2e/a2a_mesh.bats | 106 +++++++++++ tests/e2e/docker/a2a-echo/Dockerfile | 8 + tests/e2e/docker/a2a-echo/agent.py | 89 +++++++++ tests/e2e/docker/a2a-echo/client.py | 45 +++++ tests/e2e/docker/a2a-echo/requirements.txt | 5 + .../e2e/docker/a2a-echo/sam-node-config.yaml | 8 + tests/e2e/lib/container_mesh.bash | 3 +- tests/integration/a2a_test.go | 174 ++++++++++-------- 13 files changed, 466 insertions(+), 79 deletions(-) create mode 100644 tests/e2e/a2a_mesh.bats create mode 100644 tests/e2e/docker/a2a-echo/Dockerfile create mode 100644 tests/e2e/docker/a2a-echo/agent.py create mode 100644 tests/e2e/docker/a2a-echo/client.py create mode 100644 tests/e2e/docker/a2a-echo/requirements.txt create mode 100644 tests/e2e/docker/a2a-echo/sam-node-config.yaml diff --git a/charts/sam-mesh/templates/bootstrap-job.yaml b/charts/sam-mesh/templates/bootstrap-job.yaml index 1365c176..88dc086b 100644 --- a/charts/sam-mesh/templates/bootstrap-job.yaml +++ b/charts/sam-mesh/templates/bootstrap-job.yaml @@ -118,7 +118,7 @@ spec: {"name": "sam-admin", "allowed_services": ["*"], "allowed_targets": ["*"]}, {"name": "sam:role:sambox", "allowed_services": ["*"], "allowed_targets": ["*"]}, {"name": "sam:role:router", "allowed_services": ["*"], "allowed_targets": ["*"]}, - {"name": "sam:role:node", "allowed_services": {{ toJson .Values.bootstrap.nodeServices }}, "allowed_targets": ["*"]} + {"name": "sam:role:node", "allowed_services": {{ toJson .Values.bootstrap.nodeServices }}, "allowed_targets": ["*"], "allowed_labels": {{ toJson .Values.bootstrap.nodeLabels }}} ], "bindings": {{ toJson $bindings }} }' \ diff --git a/charts/sam-mesh/tests/bootstrap-job_test.yaml b/charts/sam-mesh/tests/bootstrap-job_test.yaml index 621e0dc7..f40723ac 100644 --- a/charts/sam-mesh/tests/bootstrap-job_test.yaml +++ b/charts/sam-mesh/tests/bootstrap-job_test.yaml @@ -32,3 +32,24 @@ tests: value: 100 - exists: path: spec.template.spec.containers[0].resources.limits + + - it: node role grants no labels by default (fail closed) + documentSelector: + path: kind + value: Job + asserts: + - matchRegex: + path: spec.template.spec.containers[0].args[0] + pattern: '"name": "sam:role:node".*"allowed_labels": \[\]' + + - it: node role grants only the configured label patterns + documentSelector: + path: kind + value: Job + set: + bootstrap: + nodeLabels: ["region=*", "team=platform"] + asserts: + - matchRegex: + path: spec.template.spec.containers[0].args[0] + pattern: '"name": "sam:role:node".*"allowed_labels": \["region=\*","team=platform"\]' diff --git a/charts/sam-mesh/values.yaml b/charts/sam-mesh/values.yaml index bda5d30e..de03c9ec 100644 --- a/charts/sam-mesh/values.yaml +++ b/charts/sam-mesh/values.yaml @@ -182,3 +182,6 @@ bootstrap: bindings: [] # Services sam:role:node may call nodeServices: ["*"] + # Label patterns ("key=value" or "key=*") sam:role:node may attest with + # --labels. Empty means nodes cannot enroll with any label (fail closed). + nodeLabels: [] diff --git a/internal/node/a2a_service.go b/internal/node/a2a_service.go index 22a9a808..fae5af62 100644 --- a/internal/node/a2a_service.go +++ b/internal/node/a2a_service.go @@ -91,9 +91,13 @@ const maxAgentCardBytes = 1 << 20 // the client request, fetches the card from the agent over the mesh, and // serves a regenerated card whose interfaces point at the local mesh URL. // Stock A2A clients then talk to the agent through this node unmodified. -// Non-card requests are left to the streaming egress proxy. +// The card is served both at the well-known path (resolvers that append it, +// e.g. the python SDK) and at the bare service root (resolvers that treat a +// pathful base URL as the card location, e.g. a2a-go; a root GET is not part +// of any A2A binding, JSON-RPC being POST-only). Everything else is left to +// the streaming egress proxy. func a2aServeAgentCard(node *SamNode, rt http.RoundTripper, w http.ResponseWriter, r *http.Request, route egressRoute) bool { - if r.Method != http.MethodGet || route.upstreamPath != a2aAgentCardPath { + if r.Method != http.MethodGet || (route.upstreamPath != a2aAgentCardPath && route.upstreamPath != "") { return false } resp, err := fetchRemoteAgentCard(node, rt, r, route) @@ -178,6 +182,17 @@ func regenerateAgentCardForMesh(card *a2a.AgentCard, base string) error { card.SupportedInterfaces = kept card.Capabilities.Streaming = false card.Signatures = nil + // Required list fields must stay arrays: encoding/json marshals nil + // slices as null, which strict SDK card parsers (pydantic) reject. + if card.Skills == nil { + card.Skills = []a2a.AgentSkill{} + } + if card.DefaultInputModes == nil { + card.DefaultInputModes = []string{} + } + if card.DefaultOutputModes == nil { + card.DefaultOutputModes = []string{} + } return nil } diff --git a/internal/node/a2a_service_test.go b/internal/node/a2a_service_test.go index d8f80126..5c234b1d 100644 --- a/internal/node/a2a_service_test.go +++ b/internal/node/a2a_service_test.go @@ -173,6 +173,67 @@ func TestA2AServeAgentCardRegenerates(t *testing.T) { } } +func TestA2AServeAgentCardMinimalCardStaysParseable(t *testing.T) { + // An upstream card that omits skills/defaultInputModes/defaultOutputModes: + // the regenerated card must keep them as arrays, not null, or strict SDK + // parsers (pydantic) refuse the whole card. + rt := roundTripFunc(func(*http.Request) (*http.Response, error) { + return cardResponse(http.StatusOK, `{"name":"minimal",`+ + `"supportedInterfaces":[{"url":"http://localhost:7777","protocolBinding":"JSONRPC"}],`+ + `"capabilities":{}}`), nil + }) + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json", nil) + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatal("card GET must be handled locally") + } + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200, body: %s", rec.Code, rec.Body.String()) + } + var raw map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &raw); err != nil { + t.Fatalf("regenerated card is not JSON: %v", err) + } + for _, key := range []string{"skills", "defaultInputModes", "defaultOutputModes", "supportedInterfaces"} { + if _, ok := raw[key].([]any); !ok { + t.Errorf("%s = %v (%T), must be a JSON array", key, raw[key], raw[key]) + } + } +} + +func TestA2AServeAgentCardAtServiceRoot(t *testing.T) { + // a2a-go's stock resolver treats a pathful base URL as the card URL + // itself, so the bare service root must serve the card too; the fetch + // upstream still targets the agent's well-known path. + var outbound *http.Request + rt := roundTripFunc(func(r *http.Request) (*http.Response, error) { + outbound = r + return cardResponse(http.StatusOK, + `{"name":"T","supportedInterfaces":[{"url":"http://localhost:7777","protocolBinding":"JSONRPC"}]}`), nil + }) + for _, path := range []string{"/sam/12D3KooWpeer/a2a/agent", "/sam/12D3KooWpeer/a2a/agent/"} { + rec := httptest.NewRecorder() + req := httptest.NewRequest("GET", path, nil) + req.Host = "127.0.0.1:8080" + if !serveEgressLocally(nil, rt, rec, req) { + t.Fatalf("GET %s must serve the card", path) + } + if rec.Code != http.StatusOK { + t.Fatalf("GET %s status = %d, body: %s", path, rec.Code, rec.Body.String()) + } + if want := "libp2p://12D3KooWpeer/a2a/agent/.well-known/agent-card.json"; outbound.URL.String() != want { + t.Errorf("upstream fetch URL = %q, want %q", outbound.URL.String(), want) + } + var card a2a.AgentCard + if err := json.Unmarshal(rec.Body.Bytes(), &card); err != nil { + t.Fatalf("GET %s: invalid card: %v", path, err) + } + if len(card.SupportedInterfaces) != 1 || card.SupportedInterfaces[0].URL != "http://127.0.0.1:8080/sam/12D3KooWpeer/a2a/agent" { + t.Errorf("GET %s: interfaces = %+v", path, card.SupportedInterfaces) + } + } +} + func TestA2AServeAgentCardIgnoresNonCardRequests(t *testing.T) { rt := roundTripFunc(func(*http.Request) (*http.Response, error) { t.Fatal("no mesh fetch expected") @@ -180,6 +241,7 @@ func TestA2AServeAgentCardIgnoresNonCardRequests(t *testing.T) { }) for _, tc := range []struct{ method, path string }{ {"POST", "/sam/12D3KooWpeer/a2a/agent/.well-known/agent-card.json"}, + {"POST", "/sam/12D3KooWpeer/a2a/agent"}, {"GET", "/sam/12D3KooWpeer/a2a/agent/tasks/1"}, {"GET", "/sam/12D3KooWpeer/mcp/svc/.well-known/agent-card.json"}, } { diff --git a/tests/e2e/a2a_mesh.bats b/tests/e2e/a2a_mesh.bats new file mode 100644 index 00000000..f4e17a17 --- /dev/null +++ b/tests/e2e/a2a_mesh.bats @@ -0,0 +1,106 @@ +#!/usr/bin/env bats + +load "lib/container_mesh.bash" + +A2A_ECHO_IMAGE="sam-a2a-echo:local" + +build_a2a_echo_image() { + if ! docker image inspect "${A2A_ECHO_IMAGE}" >/dev/null 2>&1; then + docker build -t "${A2A_ECHO_IMAGE}" \ + -f tests/e2e/docker/a2a-echo/Dockerfile \ + tests/e2e/docker/a2a-echo >/dev/null + fi +} + +start_a2a_echo() { + local name="${MESH_PREFIX}-a2a-echo" + docker run -d \ + --name "${name}" \ + --network "${MESH_NETWORK}" \ + --network-alias a2a-echo \ + "${A2A_ECHO_IMAGE}" >/dev/null + MESH_CONTAINERS+=("${name}") + mesh_wait_for_log "${name}" "Uvicorn running on" 30 +} + +setup() { + mesh_setup_env + build_a2a_echo_image +} + +teardown() { + mesh_cleanup_env +} + +# CUJ: bring a stock a2a-sdk agent onto the mesh via node config, then use a +# stock a2a-sdk client on another node — bootstrapping from the regenerated +# agent card — plus the fail-closed labels gate on the raw a2a egress path. +@test "a2a: stock SDK client chats with a mesh-hosted agent via the regenerated card" { + run mesh_start_mock_oidc + [[ "$status" -eq 0 ]] + + mesh_start_router + + echo "[$(date +%T)] Starting Node 1 (consumer)" + mesh_start_node 1 "--log-level debug" + mesh_wait_for_log "${MESH_PREFIX}-node-1" "SAM Node Online" 60 + mesh_wait_for_mcp_ready 1 20 + + echo "[$(date +%T)] Starting a2a echo agent backend" + start_a2a_echo + + echo "[$(date +%T)] Starting Node 2 (provider, region=eu) with the echo service" + mesh_start_node 2 \ + "--log-level debug --labels region=eu" \ + "tests/e2e/docker/a2a-echo/sam-node-config.yaml" + mesh_wait_for_log "${MESH_PREFIX}-node-2" "SAM Node Online" 20 + mesh_wait_for_mcp_ready 2 20 + + local node2_peer_id + node2_peer_id=$(docker logs "${MESH_PREFIX}-node-2" 2>&1 | grep "PeerID:" | head -n 1 | awk '{print $2}' | tr -d '\r') + + echo "[$(date +%T)] Connecting Node 1 to Node 2" + local node2_addr="/dns4/${MESH_PREFIX}-node-2/tcp/5002/p2p/${node2_peer_id}" + run mesh_connect_peer 1 "${node2_addr}" + [[ "$status" -eq 0 ]] + mesh_wait_for_peer_connection 1 "${node2_peer_id}" 20 + + local mesh_base="http://${MESH_PREFIX}-node-1:8080/sam/${node2_peer_id}/a2a/echo" + + # Stock python client: resolves the card through the mesh (client.py asserts + # the regenerated URLs, the gRPC drop and streaming-off) and gets an echo. + echo "[$(date +%T)] Running stock a2a-sdk client against ${mesh_base}" + run docker run --rm --network "${MESH_NETWORK}" \ + -e SAM_API_TOKEN="secret-token" \ + "${A2A_ECHO_IMAGE}" python3 /workspace/client.py "${mesh_base}" "hello mesh" + echo "client output: $output" + [[ "$status" -eq 0 ]] + [[ "$output" == *"agent> echo: hello mesh"* ]] + + local send_body='{"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{}}' + + # The label the provider attests (region=eu) is admitted end to end over + # the real attestation chain — same stock client, labels via header. + echo "[$(date +%T)] Labelled send (region=eu) must be admitted" + run docker run --rm --network "${MESH_NETWORK}" \ + -e SAM_API_TOKEN="secret-token" \ + -e SAM_REQUIRED_LABELS="region=eu" \ + "${A2A_ECHO_IMAGE}" python3 /workspace/client.py "${mesh_base}" "hello eu" + echo "labelled client output: $output" + [[ "$status" -eq 0 ]] + [[ "$output" == *"agent> echo: hello eu"* ]] + + # A label the provider does not attest refuses fail-closed before egress: + # the 403 comes from the caller-side gate before the body is even parsed. + echo "[$(date +%T)] Labelled send (region=us-east-1) must fail closed" + run docker run --rm --network "${MESH_NETWORK}" python:3.12 curl -s -o /dev/null -w '%{http_code}' \ + -X POST "${mesh_base}/" \ + -H "X-Sam-Authentication: Bearer secret-token" \ + -H "X-Sam-Required-Labels: region=us-east-1" \ + -H "Content-Type: application/json" \ + --max-time 30 \ + -d "${send_body}" + echo "mismatched-label status: $output" + [[ "$status" -eq 0 ]] + [[ "$output" == "403" ]] +} diff --git a/tests/e2e/docker/a2a-echo/Dockerfile b/tests/e2e/docker/a2a-echo/Dockerfile new file mode 100644 index 00000000..edfa13fa --- /dev/null +++ b/tests/e2e/docker/a2a-echo/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.12-slim + +WORKDIR /workspace +COPY ./requirements.txt /workspace/requirements.txt +RUN pip install -r requirements.txt +COPY ./agent.py ./client.py /workspace/ + +CMD ["python3", "/workspace/agent.py"] diff --git a/tests/e2e/docker/a2a-echo/agent.py b/tests/e2e/docker/a2a-echo/agent.py new file mode 100644 index 00000000..98c49fd6 --- /dev/null +++ b/tests/e2e/docker/a2a-echo/agent.py @@ -0,0 +1,89 @@ +"""Stock a2a-sdk echo agent for the e2e mesh CUJ: deterministic, no external APIs. + +The card deliberately advertises a gRPC interface and streaming so the test +can verify the mesh regenerates the card (drops gRPC, turns streaming off). +""" +import uuid + +import uvicorn +from a2a.server.agent_execution.agent_executor import AgentExecutor +from a2a.server.agent_execution.context import RequestContext +from a2a.server.events.event_queue import EventQueue +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes +from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore +from a2a.types import ( + AgentCapabilities, + AgentCard, + AgentInterface, + AgentSkill, + Message, + Part, + Role, +) +from starlette.applications import Starlette + +PORT = 7777 + + +class EchoExecutor(AgentExecutor): + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + await event_queue.enqueue_event( + Message( + role=Role.ROLE_AGENT, + message_id=str(uuid.uuid4()), + parts=[Part(text=f"echo: {context.get_user_input()}")], + context_id=context.context_id, + task_id=context.task_id, + ) + ) + + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: + pass + + +agent_card = AgentCard( + name="echo", + description="Deterministic echo agent for the a2a e2e CUJ", + version="0.1.0", + capabilities=AgentCapabilities(streaming=True), + default_input_modes=["text"], + default_output_modes=["text"], + skills=[ + AgentSkill( + id="echo", + name="echo", + description="Echoes the input back", + tags=["echo"], + ) + ], + supported_interfaces=[ + AgentInterface( + protocol_binding="JSONRPC", + protocol_version="1.0", + url=f"http://127.0.0.1:{PORT}/", + ), + # Unreachable through the mesh; the regenerated card must drop it. + AgentInterface( + protocol_binding="GRPC", + protocol_version="1.0", + url="127.0.0.1:50051", + ), + ], +) + +handler = DefaultRequestHandler( + agent_executor=EchoExecutor(), + task_store=InMemoryTaskStore(), + agent_card=agent_card, +) +# JSON-RPC at "/": the mesh card regeneration drops URL subpaths, so clients land on the root. +app = Starlette( + routes=[ + *create_jsonrpc_routes(request_handler=handler, rpc_url="/"), + *create_agent_card_routes(agent_card=agent_card), + ] +) + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=PORT) diff --git a/tests/e2e/docker/a2a-echo/client.py b/tests/e2e/docker/a2a-echo/client.py new file mode 100644 index 00000000..e9943eb5 --- /dev/null +++ b/tests/e2e/docker/a2a-echo/client.py @@ -0,0 +1,45 @@ +"""One-shot stock a2a-sdk client for the e2e CUJ: resolve, verify, send, print. + +Exits non-zero if the regenerated card is not mesh-usable or no reply arrives. +""" +import asyncio +import os +import sys +import uuid + +import httpx +from a2a.client import A2ACardResolver, ClientConfig, create_client +from a2a.helpers import get_message_text +from a2a.types import Message, Part, Role, SendMessageRequest + + +async def main(url: str, text: str) -> None: + token = os.environ.get("SAM_API_TOKEN", "secret-token") + headers = {"X-Sam-Authentication": f"Bearer {token}"} + labels = os.environ.get("SAM_REQUIRED_LABELS") + if labels: + headers["X-Sam-Required-Labels"] = labels + async with httpx.AsyncClient(timeout=60, headers=headers) as http: + card = await A2ACardResolver(http, url).get_agent_card() + bindings = [i.protocol_binding for i in card.supported_interfaces] + assert "GRPC" not in bindings, f"gRPC interface not dropped: {bindings}" + assert all(i.url == url for i in card.supported_interfaces), \ + f"interface URLs not regenerated to {url}: {card.supported_interfaces}" + assert not card.capabilities.streaming, "streaming not advertised off" + client = await create_client(card, client_config=ClientConfig(httpx_client=http)) + message = Message( + role=Role.ROLE_USER, + message_id=str(uuid.uuid4()), + parts=[Part(text=text)], + ) + async for event in client.send_message(SendMessageRequest(message=message)): + if event.HasField("message"): + print(f"agent> {get_message_text(event.message)}") + return + sys.exit("no message reply received") + + +if __name__ == "__main__": + if len(sys.argv) != 3: + sys.exit("usage: client.py ") + asyncio.run(main(sys.argv[1], sys.argv[2])) diff --git a/tests/e2e/docker/a2a-echo/requirements.txt b/tests/e2e/docker/a2a-echo/requirements.txt new file mode 100644 index 00000000..70e80861 --- /dev/null +++ b/tests/e2e/docker/a2a-echo/requirements.txt @@ -0,0 +1,5 @@ +a2a-sdk>=1.0 +httpx>=0.27 +sse-starlette>=2.0 +starlette>=0.40 +uvicorn>=0.30 diff --git a/tests/e2e/docker/a2a-echo/sam-node-config.yaml b/tests/e2e/docker/a2a-echo/sam-node-config.yaml new file mode 100644 index 00000000..0ceb2f4b --- /dev/null +++ b/tests/e2e/docker/a2a-echo/sam-node-config.yaml @@ -0,0 +1,8 @@ +version: "v1alpha1" +attenuation: + policies: [] +services: + - type: "a2a" + name: "echo" + description: "Deterministic echo agent for the a2a e2e CUJ" + target_url: "http://a2a-echo:7777" diff --git a/tests/e2e/lib/container_mesh.bash b/tests/e2e/lib/container_mesh.bash index 01d92098..3dd19273 100644 --- a/tests/e2e/lib/container_mesh.bash +++ b/tests/e2e/lib/container_mesh.bash @@ -399,7 +399,8 @@ if [[ -z "${MESH_HELPERS_LOADED:-}" ]]; then --set router.hostPort=4501 --set console.enabled=false --set 'router.externalAddrs={/dns4/sam-router/tcp/4501}' - --set 'bootstrap.nodeServices={mcp://calculator,mcp://db-agent,mcp://http-tool,mcp://stdio-tool,system://sam.catalog}') + --set 'bootstrap.nodeServices={mcp://calculator,mcp://db-agent,mcp://http-tool,mcp://stdio-tool,a2a://echo,system://sam.catalog}' + --set 'bootstrap.nodeLabels={region=*}') if ! "${helm_bin}" "${helm_args[@]}"; then # The reused cluster may hold StatefulSets whose immutable spec (e.g. # volumeClaimTemplates) changed; drop them (PVCs survive) and retry. diff --git a/tests/integration/a2a_test.go b/tests/integration/a2a_test.go index 721ea035..06ef32f2 100644 --- a/tests/integration/a2a_test.go +++ b/tests/integration/a2a_test.go @@ -16,8 +16,9 @@ package integration_test import ( "bytes" - "encoding/json" + "context" "io" + "iter" "net/http" "net/http/httptest" "path/filepath" @@ -28,13 +29,40 @@ import ( "google.golang.org/protobuf/encoding/protojson" + "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/a2asrv" "github.com/google/sam/api" ) -// TestA2ACUJ covers the "A2A agent behind the mesh" CUJ: node A (attested -// region=eu) hosts an a2a service; node B's raw egress proxy serves it with -// a rewritten agent card, admits a region=eu-labelled request, and refuses a -// region=us-east-1 request fail-closed before any payload leaves node B. +// headerRoundTripper stamps fixed headers (mesh auth, labels) on every +// request so the stock A2A SDK client needs no SAM-specific code. +type headerRoundTripper struct { + headers map[string]string +} + +func (h headerRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { + r = r.Clone(r.Context()) + for k, v := range h.headers { + r.Header.Set(k, v) + } + return http.DefaultTransport.RoundTrip(r) +} + +func meshHTTPClient(token string, extra map[string]string) *http.Client { + headers := map[string]string{api.HeaderSamAuthentication: "Bearer " + token} + for k, v := range extra { + headers[k] = v + } + return &http.Client{Transport: headerRoundTripper{headers: headers}} +} + +// TestA2ACUJ covers the "A2A agent behind the mesh" CUJ with the official +// SDK on both ends: node A (attested region=eu) hosts a stock a2asrv agent; +// a stock a2aclient bootstraps from the regenerated card served by node B, +// holds a message exchange, and a region-mismatched request is refused +// fail-closed before any payload leaves node B. func TestA2ACUJ(t *testing.T) { nodeBin := buildBinary(t, "./cmd/sam-node") _, hubAddr := startMockRouter(t) @@ -72,117 +100,113 @@ func TestA2ACUJ(t *testing.T) { } peerA := addrA[idx+len("/p2p/"):] - // Fake A2A agent on node A's side: serves its card and echoes message/send. + // Stock-SDK A2A agent on node A's side: a2asrv serving its card and + // echoing message/send. The card deliberately advertises a gRPC + // interface, streaming, and a stale signature: the mesh must drop all + // three on regeneration. var sendCount atomic.Int32 var sawLabelsHeader atomic.Bool + echo := a2asrv.AgentExecutorFunc(func(ctx context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] { + return func(yield func(a2a.Event, error) bool) { + sendCount.Add(1) + yield(a2a.NewMessageForTask(a2a.MessageRoleAgent, ec, a2a.NewTextPart("echo from eu")), nil) + } + }) + agentCard := &a2a.AgentCard{ + Name: "echo-agent", + Description: "test a2a agent", + Version: "1.0.0", + Capabilities: a2a.AgentCapabilities{Streaming: true}, + SupportedInterfaces: []*a2a.AgentInterface{ + {URL: "http://localhost:9999", ProtocolBinding: a2a.TransportProtocolJSONRPC, ProtocolVersion: "1.0"}, + {URL: "localhost:50051", ProtocolBinding: a2a.TransportProtocolGRPC, ProtocolVersion: "1.0"}, + }, + Signatures: []a2a.AgentCardSignature{{Protected: "eyJhbGciOiJFUzI1NiJ9", Signature: "c3RhbGU"}}, + } + mux := http.NewServeMux() + mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(agentCard)) + mux.Handle("/", a2asrv.NewJSONRPCHandler(a2asrv.NewHandler(echo))) agent := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Header.Get(api.HeaderSamRequiredLabels) != "" { sawLabelsHeader.Store(true) } - switch { - case r.Method == http.MethodGet && r.URL.Path == "/.well-known/agent-card.json": - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"name":"echo-agent",` + - `"supportedInterfaces":[` + - `{"url":"http://localhost:9999","protocolBinding":"JSONRPC","protocolVersion":"1.0"},` + - `{"url":"localhost:50051","protocolBinding":"GRPC","protocolVersion":"1.0"}],` + - `"capabilities":{"streaming":true}}`)) - case r.Method == http.MethodPost: - sendCount.Add(1) - w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{"kind":"message",` + - `"messageId":"m1","role":"agent","parts":[{"kind":"text","text":"echo from eu"}]}}`)) - default: - http.Error(w, "not found", http.StatusNotFound) - } + mux.ServeHTTP(w, r) })) defer agent.Close() registerA2AService(t, apiAddrA, apiToken, "echo-agent", agent.URL) meshBase := "http://" + apiAddrB + "/sam/" + peerA + "/a2a/echo-agent" + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() - // CUJ step 1: the agent card comes back rewritten for mesh use. Poll: - // the first fetch can race connectivity establishment. + // CUJ step 1: a stock SDK resolver bootstraps from the regenerated card. + // Poll: the first fetch can race connectivity establishment. + resolver := agentcard.NewResolver(meshHTTPClient(apiToken, nil)) + var card *a2a.AgentCard deadline := time.Now().Add(30 * time.Second) - var cardBody []byte for { - req, _ := http.NewRequest("GET", meshBase+"/.well-known/agent-card.json", nil) - req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) - resp, err := http.DefaultClient.Do(req) + var err error + card, err = resolver.Resolve(ctx, meshBase) if err == nil { - cardBody, _ = io.ReadAll(resp.Body) - _ = resp.Body.Close() - if resp.StatusCode == http.StatusOK { - break - } + break } if time.Now().After(deadline) { - t.Fatalf("timeout fetching agent card, last body: %s", string(cardBody)) + t.Fatalf("timeout resolving agent card through the mesh: %v", err) } time.Sleep(200 * time.Millisecond) } - var card struct { - SupportedInterfaces []struct { - URL string `json:"url"` - ProtocolBinding string `json:"protocolBinding"` - } `json:"supportedInterfaces"` - Capabilities struct { - Streaming bool `json:"streaming"` - } `json:"capabilities"` - } - if err := json.Unmarshal(cardBody, &card); err != nil { - t.Fatalf("invalid card: %v, body: %s", err, string(cardBody)) - } if len(card.SupportedInterfaces) != 1 || card.SupportedInterfaces[0].URL != meshBase { - t.Errorf("interfaces not regenerated / gRPC not dropped: %s", string(cardBody)) + t.Errorf("interfaces not regenerated / gRPC not dropped: %+v", card.SupportedInterfaces) + } + if card.SupportedInterfaces[0].ProtocolBinding != a2a.TransportProtocolJSONRPC { + t.Errorf("binding = %q, want JSONRPC", card.SupportedInterfaces[0].ProtocolBinding) } if card.Capabilities.Streaming { t.Error("streaming must be advertised off through the mesh") } + if len(card.Signatures) != 0 { + t.Error("stale signatures must be dropped from the regenerated card") + } - // CUJ step 2: message/send constrained to region=eu is admitted. - sendBody := `{"jsonrpc":"2.0","id":1,"method":"message/send","params":{"message":` + - `{"kind":"message","messageId":"c1","role":"user","parts":[{"kind":"text","text":"hi"}]}}}` + // CUJ step 2: a stock SDK client built from that card, constrained to + // region=eu, is admitted and gets the echo back. + euClient, err := a2aclient.NewFromCard(ctx, card, + a2aclient.WithJSONRPCTransport(meshHTTPClient(apiToken, map[string]string{api.HeaderSamRequiredLabels: "region=eu"}))) + if err != nil { + t.Fatalf("stock client rejected the regenerated card: %v", err) + } + req := &a2a.SendMessageRequest{Message: a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("hi"))} deadline = time.Now().Add(30 * time.Second) + var result a2a.SendMessageResult for { - req, _ := http.NewRequest("POST", meshBase+"/", strings.NewReader(sendBody)) - req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) - req.Header.Set("Content-Type", "application/json") - req.Header.Set(api.HeaderSamRequiredLabels, "region=eu") - resp, err := http.DefaultClient.Do(req) - if err != nil { - t.Fatalf("labelled message/send failed: %v", err) - } - body, _ := io.ReadAll(resp.Body) - _ = resp.Body.Close() - if resp.StatusCode == http.StatusOK { - if !strings.Contains(string(body), "echo from eu") { - t.Fatalf("unexpected message/send response: %s", string(body)) - } + result, err = euClient.SendMessage(ctx, req) + if err == nil { break } if time.Now().After(deadline) { - t.Fatalf("labelled message/send status: %d, body: %s", resp.StatusCode, string(body)) + t.Fatalf("labelled message/send failed: %v", err) } time.Sleep(200 * time.Millisecond) } + reply, ok := result.(*a2a.Message) + if !ok { + t.Fatalf("send result = %T, want *a2a.Message", result) + } + if len(reply.Parts) == 0 || reply.Parts[0].Text() != "echo from eu" { + t.Fatalf("unexpected reply: %+v", reply) + } // CUJ step 3: a mismatched label refuses fail-closed BEFORE egress — // the agent backend must never see the request. before := sendCount.Load() - req, _ := http.NewRequest("POST", meshBase+"/", strings.NewReader(sendBody)) - req.Header.Set(api.HeaderSamAuthentication, "Bearer "+apiToken) - req.Header.Set("Content-Type", "application/json") - req.Header.Set(api.HeaderSamRequiredLabels, "region=us-east-1") - respUS, err := http.DefaultClient.Do(req) + usClient, err := a2aclient.NewFromCard(ctx, card, + a2aclient.WithJSONRPCTransport(meshHTTPClient(apiToken, map[string]string{api.HeaderSamRequiredLabels: "region=us-east-1"}))) if err != nil { - t.Fatalf("mismatched-label message/send failed: %v", err) + t.Fatalf("client construction failed: %v", err) } - usBody, _ := io.ReadAll(respUS.Body) - _ = respUS.Body.Close() - if respUS.StatusCode != http.StatusForbidden { - t.Fatalf("mismatched label must fail closed with 403: got %d, body: %s", respUS.StatusCode, string(usBody)) + if _, err := usClient.SendMessage(ctx, req); err == nil { + t.Fatal("mismatched label must fail closed") } if sendCount.Load() != before { t.Fatal("payload reached the agent despite label refusal")