diff --git a/Makefile b/Makefile index e4b986ba..c21a637e 100644 --- a/Makefile +++ b/Makefile @@ -33,6 +33,9 @@ build: # nano-init is a separate module: it carries a userspace TCP stack, which # has no business in the dependency graph every other binary builds from. go -C cmd/nano-init build -v -o "$(OUT_DIR)/nano-init" . + # sam-a2a-bridge is a separate module too: it keeps the a2a SDK out of + # the root dependency graph. + go -C cmd/sam-a2a-bridge build -v -o "$(OUT_DIR)/sam-a2a-bridge" . .PHONY: mobile-ffi-host mobile-ffi-android mobile-ffi-android-x86_64 mobile-ffi-ios mobile-ffi mobile-app-apk mobile-app-apk-emulator diff --git a/agents/skills/sam-a2a-bridge/SKILL.md b/agents/skills/sam-a2a-bridge/SKILL.md new file mode 100644 index 00000000..233c69a7 --- /dev/null +++ b/agents/skills/sam-a2a-bridge/SKILL.md @@ -0,0 +1,109 @@ +--- +name: sam-a2a-bridge +description: "Use when the task should be delegated to a remote A2A agent on the SAM (Sovereign Agent Mesh) network: send it work with send_agent_task (with text, structured data, or file attachments), check agent capabilities with get_agent_card, poll results with get_agent_task, hold multi-turn conversations, and enforce data-sovereignty labels on every call. Also use to set up the sam-a2a-bridge MCP server when those tools are not callable yet." +--- + +# SAM A2A Bridge Skill + +Use this skill to delegate work to A2A agents hosted by SAM mesh peers. The +bridge exposes three tools; everything else (auth, routing, the labels +gate) happens inside the local `sam-node`. + +Pick the path that matches the need: + +- The tools are not callable yet: [Set Up The Bridge](#set-up-the-bridge). +- Before sending data or files, check what the agent accepts: + [Check Agent Capabilities](#check-agent-capabilities). +- The task needs a remote agent to do something: + [Send A Task](#send-a-task). +- A previous send returned a non-terminal state: + [Poll A Task](#poll-a-task). +- The conversation with the agent continues: + [Multi-Turn](#multi-turn). +- A call failed: [Interpret Errors](#interpret-errors). + +## Set Up The Bridge + +The bridge is a stdio MCP server that talks to the local `sam-node` sidecar. +Propose each shell command and let the user approve it before running anything. + +1. A running, enrolled `sam-node` is required first. If its sidecar does not + answer on `http://localhost:8080`, use the `sam-mesh` skill's bootstrap + path before continuing. +2. Build the bridge (own Go module; the Makefile pins the Go toolchain the + module needs, so a stale system Go still works): + `make -C /cmd/sam-a2a-bridge build` +3. Register it with the harness, passing the sidecar URL and its API token: + `claude mcp add sam-a2a-bridge -- /bin/sam-a2a-bridge -url http://localhost:8080 -token ` + (Agent-returned files land in `~/.sam/a2a-downloads` by default, auto-created; pass `-download-dir` to change it — you must create that directory yourself. No auto-cleanup; prune manually.) +4. Restart the harness session; the three tools appear. + +## Check Agent Capabilities + +`get_agent_card(peer, service)` + +Call this before composing structured data or file attachments to verify what +the agent accepts. Returns a trimmed card including: +- Registered skills with examples +- `default_input_modes` — accepted MIME types for structured data (e.g., + `application/json`) +- `default_output_modes` — MIME types the agent can return +- Whether the agent supports streaming + +Use the capabilities to shape your `data` (JSON object matching the agent's +input schema) and `file_path` attachments appropriately. + +## Send A Task + +`send_agent_task(peer, service, message?, data?, file_path?, file_name?, required_labels?, context_id?, task_id?)` + +- `peer` is the provider node's peer ID; `service` is the a2a service name it + registered. If unknown, discover them with the `sam-node` MCP tool + `discover_remote_services` with `type: a2a`, or ask the user. +- At least one of `message`, `data`, or `file_path` is required. All are combinable. + - `message` is plain text. + - `data` is a JSON object; it becomes a DataPart and routes to the agent's + structured-input handler. Check `default_input_modes` first. + - `file_path` attaches a local file (up to 5 MB); `file_name` optionally + renames what the agent sees (default: the file's base name). +- The call returns immediately with `{"task_id", "context_id", "state", "text", + "data"?, "files"?}` — it never blocks on the agent. + - `data` (optional) contains structured JSON returned inline. + - `files` (optional) is a list of paths where agent-returned files are saved + under the download directory. +- **Sovereignty**: when the task involves data that must stay in a region or + jurisdiction, set `required_labels` (comma-separated `key=value`, e.g. + `region=eu-west-1`). The local node then refuses fail-closed before any data + leaves it unless the peer's control-plane-attested labels match. Never drop + or weaken `required_labels` to make a refused call succeed without the + user's explicit approval — the refusal is the feature. + +## Poll A Task + +If `state` is not terminal (`completed`, `failed`, `canceled`, `rejected`), +poll with `get_agent_task(peer, service, task_id)` until it is. Space polls a +few seconds apart; agent tasks can be slow. `text` carries the agent's status +message while running and its answer or artifacts when completed. `data` and +`files` appear only when present. + +## Multi-Turn + +- Follow-up question in the same conversation: pass the returned `context_id` + on the next `send_agent_task`. Without it every message is a cold start. +- The task is in state `input-required` (the agent asked something): answer by + passing BOTH `task_id` and `context_id` — that routes the reply into the + waiting task so it can finish. Terminal tasks cannot receive messages. + +## Interpret Errors + +- `403: Required labels not attested by provider` — the sovereignty gate + refused before egress. Expected for non-matching regions; report it to the + user, do not retry with weaker labels on your own. +- `400: Invalid X-Sam-Required-Labels header ...` — malformed labels; fix the + `key=value,key=value` syntax. +- `404` / `Service not found` — the peer has no a2a service by that name; + re-discover or check the name with the user. +- Connection refused / timeout — the local sidecar URL or token is wrong, or + the node is down; go to [Set Up The Bridge](#set-up-the-bridge). +- Interop note: the remote agent must run a2a-go v2.x; older A2A stacks speak + a different JSON-RPC dialect and will not answer. diff --git a/cmd/sam-a2a-bridge/.gitignore b/cmd/sam-a2a-bridge/.gitignore new file mode 100644 index 00000000..5af00c72 --- /dev/null +++ b/cmd/sam-a2a-bridge/.gitignore @@ -0,0 +1 @@ +/sam-a2a-bridge diff --git a/cmd/sam-a2a-bridge/Makefile b/cmd/sam-a2a-bridge/Makefile new file mode 100644 index 00000000..ef8e9756 --- /dev/null +++ b/cmd/sam-a2a-bridge/Makefile @@ -0,0 +1,28 @@ +# Pinned because a stale system Go (1.24) fails to auto-resolve the +# toolchain go.mod needs; override with GOTOOLCHAIN=auto if yours does. +GOTOOLCHAIN ?= go1.25.7 +export GOTOOLCHAIN + +BINARY := sam-a2a-bridge +# Binaries land in the repo-root bin/ like every other sam binary. +OUT_DIR := $(abspath ../../bin) + +.PHONY: build test vet fmt install clean + +build: + go build -o "$(OUT_DIR)/$(BINARY)" . + +test: + go test ./... + +vet: + go vet ./... + +fmt: + gofmt -l . + +install: + go install . + +clean: + rm -f "$(OUT_DIR)/$(BINARY)" diff --git a/cmd/sam-a2a-bridge/a2a.go b/cmd/sam-a2a-bridge/a2a.go new file mode 100644 index 00000000..0964998a --- /dev/null +++ b/cmd/sam-a2a-bridge/a2a.go @@ -0,0 +1,356 @@ +// 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 main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "mime" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" + + "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/a2aproject/a2a-go/v2/a2aclient" +) + +const maxAttachmentBytes = 5 << 20 + +type bridgeConfig struct { + sidecarURL string + token string + downloadDir string +} + +// meshURL is the sidecar's raw egress path for one remote a2a service. +func (c bridgeConfig) meshURL(peer, service string) string { + return strings.TrimRight(c.sidecarURL, "/") + "/sam/" + url.PathEscape(peer) + "/a2a/" + url.PathEscape(service) +} + +type getAgentCardParams struct { + Peer string `json:"peer" jsonschema:"Peer ID of the node hosting the agent"` + Service string `json:"service" jsonschema:"Name of the a2a service registered on that peer"` +} + +// agentCardSummary trims the agent card to what a model can use; full cards +// carry security schemas and provider blurbs a model never needs. +type agentCardSummary struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + Version string `json:"version,omitempty"` + DefaultInputModes []string `json:"default_input_modes,omitempty"` + DefaultOutputModes []string `json:"default_output_modes,omitempty"` + Streaming bool `json:"streaming"` + Skills []agentCardSkill `json:"skills,omitempty"` +} + +type agentCardSkill struct { + ID string `json:"id"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` + Tags []string `json:"tags,omitempty"` + Examples []string `json:"examples,omitempty"` + InputModes []string `json:"input_modes,omitempty"` + OutputModes []string `json:"output_modes,omitempty"` +} + +func handleGetAgentCard(ctx context.Context, cfg bridgeConfig, p getAgentCardParams) (agentCardSummary, error) { + httpClient := &http.Client{ + Timeout: 30 * time.Second, + Transport: &samTransport{base: http.DefaultTransport, token: cfg.token}, + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, + cfg.meshURL(p.Peer, p.Service)+"/.well-known/agent-card.json", nil) + if err != nil { + return agentCardSummary{}, err + } + resp, err := httpClient.Do(req) + if err != nil { + return agentCardSummary{}, err + } + defer func() { _ = resp.Body.Close() }() + + var card a2a.AgentCard + if err := json.NewDecoder(resp.Body).Decode(&card); err != nil { + return agentCardSummary{}, fmt.Errorf("agent card is not valid JSON: %w", err) + } + summary := agentCardSummary{ + Name: card.Name, + Description: card.Description, + Version: card.Version, + DefaultInputModes: card.DefaultInputModes, + DefaultOutputModes: card.DefaultOutputModes, + Streaming: card.Capabilities.Streaming, + } + for _, skill := range card.Skills { + summary.Skills = append(summary.Skills, agentCardSkill{ + ID: skill.ID, + Name: skill.Name, + Description: skill.Description, + Tags: skill.Tags, + Examples: skill.Examples, + InputModes: skill.InputModes, + OutputModes: skill.OutputModes, + }) + } + return summary, nil +} + +type sendAgentTaskParams struct { + Peer string `json:"peer" jsonschema:"Peer ID of the node hosting the agent"` + Service string `json:"service" jsonschema:"Name of the a2a service registered on that peer"` + Message string `json:"message,omitempty" jsonschema:"Plain-text message for the agent; optional if data or file_path is set"` + Data map[string]any `json:"data,omitempty" jsonschema:"Structured JSON payload sent to the agent as an A2A DataPart"` + FilePath string `json:"file_path,omitempty" jsonschema:"Local file to attach; sent to the agent as bytes, max 5 MB"` + FileName string `json:"file_name,omitempty" jsonschema:"Name shown to the agent for the attached file (default: the file's base name)"` + RequiredLabels string `json:"required_labels,omitempty" jsonschema:"Comma-separated key=value labels the provider must have attested (e.g. region=eu-west-1); the local node refuses fail-closed before any data leaves it"` + ContextID string `json:"context_id,omitempty" jsonschema:"Continue an existing conversation context"` + TaskID string `json:"task_id,omitempty" jsonschema:"Reply into an existing task, e.g. one in state input-required"` +} + +type taskResult struct { + TaskID string `json:"task_id"` + ContextID string `json:"context_id"` + State string `json:"state"` + Text string `json:"text"` + Data []any `json:"data,omitempty"` + Files []string `json:"files,omitempty"` +} + +func handleSendAgentTask(ctx context.Context, cfg bridgeConfig, p sendAgentTaskParams) (taskResult, error) { + client, err := newMeshClient(ctx, cfg, p.Peer, p.Service, p.RequiredLabels) + if err != nil { + return taskResult{}, err + } + defer client.Destroy() + + parts, err := buildParts(p) + if err != nil { + return taskResult{}, err + } + var msg *a2a.Message + if p.TaskID != "" || p.ContextID != "" { + msg = a2a.NewMessageForTask(a2a.MessageRoleUser, + a2a.TaskInfo{TaskID: a2a.TaskID(p.TaskID), ContextID: p.ContextID}, parts...) + } else { + msg = a2a.NewMessage(a2a.MessageRoleUser, parts...) + } + result, err := client.SendMessage(ctx, &a2a.SendMessageRequest{Message: msg}) + if err != nil { + return taskResult{}, err + } + return toTaskResult(cfg, result) +} + +type getAgentTaskParams struct { + Peer string `json:"peer" jsonschema:"Peer ID of the node hosting the agent"` + Service string `json:"service" jsonschema:"Name of the a2a service registered on that peer"` + TaskID string `json:"task_id" jsonschema:"ID of the task to fetch"` +} + +func handleGetAgentTask(ctx context.Context, cfg bridgeConfig, p getAgentTaskParams) (taskResult, error) { + client, err := newMeshClient(ctx, cfg, p.Peer, p.Service, "") + if err != nil { + return taskResult{}, err + } + defer client.Destroy() + + task, err := client.GetTask(ctx, &a2a.GetTaskRequest{ID: a2a.TaskID(p.TaskID)}) + if err != nil { + return taskResult{}, err + } + return toTaskResult(cfg, task) +} + +func newMeshClient(ctx context.Context, cfg bridgeConfig, peer, service, requiredLabels string) (*a2aclient.Client, error) { + httpClient := &http.Client{ + Timeout: 60 * time.Second, + Transport: &samTransport{ + base: http.DefaultTransport, + token: cfg.token, + requiredLabels: requiredLabels, + }, + } + return a2aclient.NewFromEndpoints(ctx, + []*a2a.AgentInterface{a2a.NewAgentInterface(cfg.meshURL(peer, service), a2a.TransportProtocolJSONRPC)}, + a2aclient.WithJSONRPCTransport(httpClient), + ) +} + +// buildParts builds the message parts in a fixed order (text, data, file) so +// wire output is deterministic across calls with the same params. +func buildParts(p sendAgentTaskParams) ([]*a2a.Part, error) { + var parts []*a2a.Part + if p.Message != "" { + parts = append(parts, a2a.NewTextPart(p.Message)) + } + if p.Data != nil { + parts = append(parts, a2a.NewDataPart(p.Data)) + } + if p.FilePath != "" { + filePart, err := fileToPart(p.FilePath, p.FileName) + if err != nil { + return nil, err + } + parts = append(parts, filePart) + } + if len(parts) == 0 { + return nil, fmt.Errorf("nothing to send: set message, data, or file_path") + } + return parts, nil +} + +// fileToPart reads path into a raw Part; the SDK's Part has no dedicated file +// constructor, so filename/mediaType are set directly on the returned Part. +func fileToPart(path, nameOverride string) (*a2a.Part, error) { + info, err := os.Stat(path) + if err != nil { + return nil, err + } + // A FIFO or device would block or misbehave in ReadFile. + if !info.Mode().IsRegular() { + return nil, fmt.Errorf("file %s is not a regular file", path) + } + if info.Size() > maxAttachmentBytes { + return nil, fmt.Errorf("file %s is %d bytes; attachment cap is %d", path, info.Size(), maxAttachmentBytes) + } + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + name := nameOverride + if name == "" { + name = filepath.Base(path) + } + mimeType := mime.TypeByExtension(filepath.Ext(name)) + if mimeType == "" { + mimeType = http.DetectContentType(data) + } + part := a2a.NewRawPart(data) + part.Filename = name + part.MediaType = mimeType + return part, nil +} + +// toTaskResult flattens the SDK's Message|Task union into the fields a +// harness needs, also collecting data/file parts from every location; +// a direct Message reply is final, hence state "completed". +func toTaskResult(cfg bridgeConfig, result any) (taskResult, error) { + switch v := result.(type) { + case *a2a.Message: + out := taskResult{ + TaskID: string(v.TaskID), + ContextID: v.ContextID, + State: "completed", + Text: textOf(v.Parts), + } + if err := out.collect(cfg, out.TaskID, v.Parts); err != nil { + return taskResult{}, err + } + return out, nil + case *a2a.Task: + out := taskResult{TaskID: string(v.ID), ContextID: v.ContextID, State: string(v.Status.State)} + if v.Status.Message != nil { + out.Text = textOf(v.Status.Message.Parts) + if err := out.collect(cfg, out.TaskID, v.Status.Message.Parts); err != nil { + return taskResult{}, err + } + } + if out.Text == "" { + var texts []string + for _, artifact := range v.Artifacts { + if s := textOf(artifact.Parts); s != "" { + texts = append(texts, s) + } + } + out.Text = strings.Join(texts, "\n") + } + for _, artifact := range v.Artifacts { + if err := out.collect(cfg, out.TaskID, artifact.Parts); err != nil { + return taskResult{}, err + } + } + return out, nil + } + return taskResult{}, nil +} + +// textOf joins the text of every part; a2a.Part is a concrete struct (not an +// interface) so Text() is universal, unlike the docs-only sketch assumed. +func textOf(parts a2a.ContentParts) string { + var texts []string + for _, part := range parts { + if s := part.Text(); s != "" { + texts = append(texts, s) + } + } + return strings.Join(texts, "\n") +} + +// collect appends data parts inline and saves file parts, returning paths. +// Paths, not base64: inline bytes flood the model's context; a path the +// model can act on is the useful representation. +func (r *taskResult) collect(cfg bridgeConfig, taskID string, parts a2a.ContentParts) error { + for _, part := range parts { + if d := part.Data(); d != nil { + r.Data = append(r.Data, d) + } + if uri := part.URL(); uri != "" { + r.Files = append(r.Files, string(uri)) + continue + } + if b := part.Raw(); b != nil { + path, err := saveFilePart(cfg.downloadDir, taskID, part.Filename, b) + if err != nil { + return fmt.Errorf("task %s: saving returned file: %w", taskID, err) + } + r.Files = append(r.Files, path) + } + } + return nil +} + +func saveFilePart(dir, taskID, name string, data []byte) (string, error) { + if taskID == "" { + taskID = "msg" + } + if name == "" { + name = "file" + } + // Both components are remote input naming a local path: keep only leaves. + taskID = filepath.Base(filepath.Clean("/" + taskID)) + name = filepath.Base(filepath.Clean("/" + name)) + base := taskID + "-" + name + path := filepath.Join(dir, base) + for n := 1; ; n++ { + _, err := os.Stat(path) + if errors.Is(err, os.ErrNotExist) { + break + } + if err != nil { + return "", err + } + path = filepath.Join(dir, fmt.Sprintf("%s.%d", base, n)) + } + if err := os.WriteFile(path, data, 0o644); err != nil { + return "", err + } + return path, nil +} diff --git a/cmd/sam-a2a-bridge/a2a_test.go b/cmd/sam-a2a-bridge/a2a_test.go new file mode 100644 index 00000000..236bb841 --- /dev/null +++ b/cmd/sam-a2a-bridge/a2a_test.go @@ -0,0 +1,393 @@ +// 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 main + +import ( + "context" + "encoding/base64" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +// fakeSidecar records the last JSON-RPC request and returns a canned result. +func fakeSidecar(t *testing.T, wantPath string, result string, capture *map[string]any) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, wantPath) { + t.Errorf("request path = %q, want prefix %q", r.URL.Path, wantPath) + } + body, _ := io.ReadAll(r.Body) + var rpc map[string]any + _ = json.Unmarshal(body, &rpc) + if capture != nil { + *capture = rpc + } + w.Header().Set("Content-Type", "application/json") + id, _ := json.Marshal(rpc["id"]) + _, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":` + string(id) + `,"result":` + result + `}`)) + })) +} + +// SendMessage results arrive enveloped as {"task":...}/{"message":...} per +// a2a.StreamResponse; GetTask results don't (unmarshaled directly into a2a.Task). + +func TestSendAgentTaskMapsTaskResult(t *testing.T) { + var rpc map[string]any + task := `{"task":{"kind":"task","id":"t1","contextId":"c1",` + + `"status":{"state":"working","message":{"kind":"message","messageId":"m1","role":"agent",` + + `"parts":[{"kind":"text","text":"thinking"}]}}}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", task, &rpc) + defer srv.Close() + + cfg := bridgeConfig{sidecarURL: srv.URL, token: "tok"} + got, err := handleSendAgentTask(context.Background(), cfg, sendAgentTaskParams{ + Peer: "12D3KooWpeer", Service: "echo", Message: "hi", + }) + if err != nil { + t.Fatal(err) + } + if got.TaskID != "t1" || got.ContextID != "c1" { + t.Errorf("ids = %q/%q, want t1/c1", got.TaskID, got.ContextID) + } + if !strings.Contains(strings.ToLower(got.State), "working") { + t.Errorf("state = %q, want it to convey 'working'", got.State) + } + if got.Text != "thinking" { + t.Errorf("text = %q, want %q", got.Text, "thinking") + } + // a2a-go v2.5.0 sends the JSON-RPC method as "SendMessage" (internal/jsonrpc.MethodMessageSend), + // not the wire-spec "message/send". + if m, _ := rpc["method"].(string); m != "SendMessage" { + t.Errorf("jsonrpc method = %q, want SendMessage", m) + } +} + +func TestSendAgentTaskMapsMessageResult(t *testing.T) { + msg := `{"message":{"kind":"message","messageId":"m2","role":"agent","taskId":"t2","contextId":"c2",` + + `"parts":[{"kind":"text","text":"4"}]}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", msg, nil) + defer srv.Close() + + got, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "2+2?"}) + if err != nil { + t.Fatal(err) + } + if got.Text != "4" || got.TaskID != "t2" || got.ContextID != "c2" { + t.Errorf("got %+v", got) + } + if got.State != "completed" { + t.Errorf("a direct Message reply is final; state = %q, want completed", got.State) + } +} + +func TestSendAgentTaskThreadsContinuationIDs(t *testing.T) { + var rpc map[string]any + task := `{"task":{"kind":"task","id":"t3","contextId":"c3","status":{"state":"completed"}}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", task, &rpc) + defer srv.Close() + + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "more", + ContextID: "c3", TaskID: "t3"}) + if err != nil { + t.Fatal(err) + } + params, _ := rpc["params"].(map[string]any) + message, _ := params["message"].(map[string]any) + if message["taskId"] != "t3" || message["contextId"] != "c3" { + t.Errorf("continuation ids not threaded: %v", message) + } +} + +func TestSendAgentTaskSurfacesSidecarRefusal(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Required labels not attested by provider", http.StatusForbidden) + })) + defer srv.Close() + + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "hi", + RequiredLabels: "region=us-east-1"}) + if err == nil { + t.Fatal("403 must surface as an error") + } + if !strings.Contains(err.Error(), "Required labels not attested by provider") { + t.Errorf("refusal text lost: %v", err) + } +} + +func TestGetAgentTaskMapsResult(t *testing.T) { + var rpc map[string]any + task := `{"kind":"task","id":"t4","contextId":"c4","status":{"state":"completed"},` + + `"artifacts":[{"artifactId":"a1","parts":[{"kind":"text","text":"final answer"}]}]}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", task, &rpc) + defer srv.Close() + + got, err := handleGetAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + getAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", TaskID: "t4"}) + if err != nil { + t.Fatal(err) + } + // a2a-go v2.5.0 sends the JSON-RPC method as "GetTask" (internal/jsonrpc.MethodTasksGet), + // not the wire-spec "tasks/get". + if m, _ := rpc["method"].(string); m != "GetTask" { + t.Errorf("jsonrpc method = %q, want GetTask", m) + } + if got.TaskID != "t4" || got.Text != "final answer" { + t.Errorf("got %+v", got) + } + if !strings.Contains(strings.ToLower(got.State), "completed") { + t.Errorf("state = %q, want it to convey 'completed'", got.State) + } +} + +func TestMeshURLEscapesPathSegments(t *testing.T) { + cfg := bridgeConfig{sidecarURL: "http://localhost:8080"} + got := cfg.meshURL("12D3KooWpeer", "../../sam/service/register") + want := "http://localhost:8080/sam/12D3KooWpeer/a2a/..%2F..%2Fsam%2Fservice%2Fregister" + if got != want { + t.Fatalf("meshURL = %q, want %q", got, want) + } +} + +func TestSendAgentTaskWithDataPart(t *testing.T) { + var rpc map[string]any + task := `{"kind":"task","id":"t10","contextId":"c10","status":{"state":"completed"}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", `{"task":`+task+`}`, &rpc) + defer srv.Close() + + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", + Data: map[string]any{"answer": 42, "unit": "cm"}}) + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(rpc) + if !strings.Contains(string(raw), `"answer":42`) || !strings.Contains(string(raw), `"unit":"cm"`) { + t.Fatalf("data payload not on the wire: %s", raw) + } +} + +func TestSendAgentTaskWithFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "hello.txt") + if err := os.WriteFile(path, []byte("hello agent"), 0o644); err != nil { + t.Fatal(err) + } + var rpc map[string]any + task := `{"kind":"task","id":"t11","contextId":"c11","status":{"state":"completed"}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", `{"task":`+task+`}`, &rpc) + defer srv.Close() + + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", FilePath: path}) + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(rpc) + wantB64 := base64.StdEncoding.EncodeToString([]byte("hello agent")) + if !strings.Contains(string(raw), wantB64) { + t.Fatalf("file bytes not on the wire as base64: %s", raw) + } + if !strings.Contains(string(raw), "hello.txt") { + t.Fatalf("file name not on the wire: %s", raw) + } +} + +func TestSendAgentTaskFileTooLarge(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "big.bin") + if err := os.WriteFile(path, make([]byte, maxAttachmentBytes+1), 0o644); err != nil { + t.Fatal(err) + } + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: "http://127.0.0.1:1"}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", FilePath: path}) + if err == nil || !strings.Contains(err.Error(), "attachment cap") { + t.Fatalf("oversized file must be rejected before any request, got: %v", err) + } +} + +func TestSendAgentTaskRequiresContent(t *testing.T) { + _, err := handleSendAgentTask(context.Background(), bridgeConfig{sidecarURL: "http://127.0.0.1:1"}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo"}) + if err == nil || !strings.Contains(err.Error(), "nothing to send") { + t.Fatalf("empty send must be rejected, got: %v", err) + } +} + +func TestResultCollectsDataAndFiles(t *testing.T) { + downloads := t.TempDir() + fileB64 := base64.StdEncoding.EncodeToString([]byte("report body")) + task := `{"kind":"task","id":"t20","contextId":"c20","status":{"state":"completed",` + + `"message":{"kind":"message","messageId":"m1","role":"agent","parts":[` + + `{"text":"done"},{"data":{"anomalies":2}}]}},` + + `"artifacts":[{"artifactId":"a1","name":"report","parts":[` + + `{"raw":"` + fileB64 + `","filename":"report.txt","mediaType":"text/plain"},` + + `{"data":{"rows":14002}}]}]}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", `{"task":`+task+`}`, nil) + defer srv.Close() + + got, err := handleSendAgentTask(context.Background(), + bridgeConfig{sidecarURL: srv.URL, downloadDir: downloads}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "go"}) + if err != nil { + t.Fatal(err) + } + if got.Text != "done" { + t.Errorf("text = %q, want done (text logic unchanged)", got.Text) + } + if len(got.Data) != 2 { + t.Fatalf("data = %v, want 2 entries (status message + artifact)", got.Data) + } + if len(got.Files) != 1 { + t.Fatalf("files = %v, want 1", got.Files) + } + if !strings.HasPrefix(filepath.Base(got.Files[0]), "t20-") { + t.Errorf("file name %q not task-id-prefixed", got.Files[0]) + } + content, err := os.ReadFile(got.Files[0]) + if err != nil || string(content) != "report body" { + t.Fatalf("saved file wrong: %v %q", err, content) + } +} + +func TestSaveFilePartSanitizesName(t *testing.T) { + dir := t.TempDir() + path, err := saveFilePart(dir, "t1", "../../evil.txt", []byte("x")) + if err != nil { + t.Fatal(err) + } + if filepath.Dir(path) != dir { + t.Fatalf("escaped the download dir: %s", path) + } +} + +func TestSaveFilePartCollision(t *testing.T) { + dir := t.TempDir() + first, err := saveFilePart(dir, "t1", "a.txt", []byte("1")) + if err != nil { + t.Fatal(err) + } + second, err := saveFilePart(dir, "t1", "a.txt", []byte("2")) + if err != nil { + t.Fatal(err) + } + if first == second { + t.Fatalf("collision not uniquified: %s", second) + } +} + +func TestResultIncludesFileURIs(t *testing.T) { + task := `{"kind":"task","id":"t21","contextId":"c21","status":{"state":"completed"},` + + `"artifacts":[{"artifactId":"a1","parts":[` + + `{"url":"https://example.com/big.bin","mediaType":"application/octet-stream"}]}]}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", `{"task":`+task+`}`, nil) + defer srv.Close() + + got, err := handleSendAgentTask(context.Background(), + bridgeConfig{sidecarURL: srv.URL, downloadDir: t.TempDir()}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "go"}) + if err != nil { + t.Fatal(err) + } + if len(got.Files) != 1 || got.Files[0] != "https://example.com/big.bin" { + t.Fatalf("URI file part must pass through as-is: %v", got.Files) + } +} + +func TestGetAgentCardTrims(t *testing.T) { + card := `{"name":"echo-agent","description":"echoes","version":"1.2.0",` + + `"defaultInputModes":["text/plain","application/pdf"],"defaultOutputModes":["text/plain"],` + + `"capabilities":{"streaming":false,"pushNotifications":true},` + + `"securitySchemes":{"corp":{"apiKeySecurityScheme":{"location":"cookie","name":"session"}}},` + + `"provider":{"organization":"acme"},` + + `"skills":[{"id":"echo","name":"Echo","description":"repeats input",` + + `"tags":["test"],"examples":["say hi"],"inputModes":["audio/mpeg"],"outputModes":["text/plain"]}]}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || !strings.HasSuffix(r.URL.Path, "/.well-known/agent-card.json") { + http.Error(w, "not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(card)) + })) + defer srv.Close() + + got, err := handleGetAgentCard(context.Background(), bridgeConfig{sidecarURL: srv.URL}, + getAgentCardParams{Peer: "12D3KooWpeer", Service: "echo"}) + if err != nil { + t.Fatal(err) + } + if got.Name != "echo-agent" || got.Version != "1.2.0" { + t.Errorf("identity fields: %+v", got) + } + if len(got.Skills) != 1 || got.Skills[0].ID != "echo" || got.Skills[0].Examples[0] != "say hi" || + len(got.Skills[0].InputModes) != 1 || got.Skills[0].InputModes[0] != "audio/mpeg" { + t.Errorf("skills not carried: %+v", got.Skills) + } + if len(got.DefaultInputModes) != 2 { + t.Errorf("input modes not carried: %+v", got.DefaultInputModes) + } + if got.Streaming { + t.Error("streaming must reflect the card (false)") + } + raw, _ := json.Marshal(got) + if strings.Contains(string(raw), "session") || strings.Contains(string(raw), "acme") { + t.Fatalf("trim failed, security/provider material leaked: %s", raw) + } +} + +func TestResultErrorIncludesTaskID(t *testing.T) { + fileB64 := base64.StdEncoding.EncodeToString([]byte("file content")) + task := `{"kind":"task","id":"t99","contextId":"c99","status":{"state":"completed",` + + `"message":{"kind":"message","messageId":"m1","role":"agent","parts":[` + + `{"raw":"` + fileB64 + `","filename":"test.txt","mediaType":"text/plain"}]}}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", `{"task":`+task+`}`, nil) + defer srv.Close() + + _, err := handleSendAgentTask(context.Background(), + bridgeConfig{sidecarURL: srv.URL, downloadDir: "/nonexistent-dir-xyz"}, + sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "go"}) + if err == nil { + t.Fatal("error expected when download dir does not exist") + } + if !strings.Contains(err.Error(), "task t99") { + t.Errorf("error must include task ID; got: %v", err) + } +} + +func TestSaveFilePartSanitizesTaskID(t *testing.T) { + dir := t.TempDir() + path, err := saveFilePart(dir, "../../evil", "a.txt", []byte("x")) + if err != nil { + t.Fatal(err) + } + if filepath.Dir(path) != dir { + t.Fatalf("task id escaped the download dir: %s", path) + } +} + +func TestFileToPartRejectsNonRegular(t *testing.T) { + if _, err := fileToPart(t.TempDir(), ""); err == nil || !strings.Contains(err.Error(), "not a regular file") { + t.Fatalf("directory must be rejected, got: %v", err) + } +} diff --git a/cmd/sam-a2a-bridge/go.mod b/cmd/sam-a2a-bridge/go.mod new file mode 100644 index 00000000..f6866057 --- /dev/null +++ b/cmd/sam-a2a-bridge/go.mod @@ -0,0 +1,21 @@ +module github.com/google/sam/cmd/sam-a2a-bridge + +go 1.25.7 + +require ( + github.com/a2aproject/a2a-go/v2 v2.5.0 + github.com/modelcontextprotocol/go-sdk v1.7.0 +) + +require ( + github.com/google/jsonschema-go v0.4.3 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/segmentio/asm v1.1.3 // indirect + github.com/segmentio/encoding v0.5.4 // indirect + github.com/yosida95/uritemplate/v3 v3.0.2 // indirect + golang.org/x/mod v0.35.0 // indirect + golang.org/x/oauth2 v0.35.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.45.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/cmd/sam-a2a-bridge/go.sum b/cmd/sam-a2a-bridge/go.sum new file mode 100644 index 00000000..f039c34b --- /dev/null +++ b/cmd/sam-a2a-bridge/go.sum @@ -0,0 +1,30 @@ +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/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= +github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+DQPd0= +github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/modelcontextprotocol/go-sdk v1.7.0 h1:yqjY2dsbKAC0LSuWZVBMrHgiG8ukXv6NRo0JiALay44= +github.com/modelcontextprotocol/go-sdk v1.7.0/go.mod h1:dL7u98E/zjJTGzEq+j30jQ8K2k1mb6LeAH4inEcSGts= +github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= +github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= +github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= +github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= +golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= +golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= +golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= +golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.43.0 h1:12BdW9CeB3Z+J/I/wj34VMl8X+fEXBxVR90JeMX5E7s= +golang.org/x/tools v0.43.0/go.mod h1:uHkMso649BX2cZK6+RpuIPXS3ho2hZo4FVwfoy1vIk0= diff --git a/cmd/sam-a2a-bridge/main.go b/cmd/sam-a2a-bridge/main.go new file mode 100644 index 00000000..44486429 --- /dev/null +++ b/cmd/sam-a2a-bridge/main.go @@ -0,0 +1,118 @@ +// 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 main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "log" + "os" + "path/filepath" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +func main() { + sidecarURL := flag.String("url", "http://localhost:8080", "Sidecar API base URL") + token := flag.String("token", "", "Authorization Bearer token for protected sidecar endpoints") + downloadDir := flag.String("download-dir", "", "Directory for files returned by agents (default: ~/.sam/a2a-downloads, created if missing)") + flag.Parse() + + // A user-supplied dir is used as-is (the caller creates it); only the + // built-in default is auto-created. Fail fast so a missing dir surfaces + // at startup, not on the first file-bearing result. + dir := *downloadDir + if dir == "" { + home, err := os.UserHomeDir() + if err != nil { + log.Fatal(err) + } + dir = filepath.Join(home, ".sam", "a2a-downloads") + if err := os.MkdirAll(dir, 0o755); err != nil { + log.Fatal(err) + } + } else if info, err := os.Stat(dir); err != nil || !info.IsDir() { + log.Fatalf("-download-dir %s is not a usable directory (create it first): %v", dir, err) + } + cfg := bridgeConfig{sidecarURL: *sidecarURL, token: *token, downloadDir: dir} + if err := newBridgeServer(cfg).Run(context.Background(), &mcp.StdioTransport{}); err != nil { + log.Fatal(err) + } +} + +func newBridgeServer(cfg bridgeConfig) *mcp.Server { + server := mcp.NewServer(&mcp.Implementation{ + Name: "sam-a2a-bridge", + Version: "0.1.0", + }, nil) + + mcp.AddTool(server, &mcp.Tool{ + Name: "get_agent_card", + Description: "Fetch a mesh agent's card, trimmed to essentials: skills with examples, accepted " + + "input/output MIME types, and streaming support. Use before composing data/file sends.", + }, mcpTool(cfg, handleGetAgentCard)) + + mcp.AddTool(server, &mcp.Tool{ + Name: "send_agent_task", + Description: "Send a message to an A2A agent on the SAM mesh: plain text (message), structured JSON " + + "(data), and/or a local file attachment (file_path, max 5 MB) — at least one required. Returns " + + "immediately with {task_id, context_id, state, text, data?, files?}; received files are saved to " + + "disk and returned as paths. Poll non-terminal states with get_agent_task. Pass context_id to " + + "continue a conversation, task_id to reply into a task waiting for input. required_labels makes " + + "the local node refuse fail-closed unless the provider attested them.", + }, mcpTool(cfg, handleSendAgentTask)) + + mcp.AddTool(server, &mcp.Tool{ + Name: "get_agent_task", + Description: "Fetch the current state and output of a task previously created with send_agent_task; " + + "same result shape, including data and files.", + }, mcpTool(cfg, handleGetAgentTask)) + + return server +} + +// mcpTool adapts a handler to the go-sdk tool signature, routing errors +// through toolResult so refusals surface as tool errors. +func mcpTool[Params, Result any](cfg bridgeConfig, handler func(context.Context, bridgeConfig, Params) (Result, error)) func(context.Context, *mcp.CallToolRequest, Params) (*mcp.CallToolResult, any, error) { + return func(ctx context.Context, _ *mcp.CallToolRequest, p Params) (*mcp.CallToolResult, any, error) { + res, err := handler(ctx, cfg, p) + return toolResult(res, err) + } +} + +// toolResult renders errors as tool errors (not protocol errors) so the +// harness model sees refusal text like the labels-gate 403 verbatim. +func toolResult(res any, err error) (*mcp.CallToolResult, any, error) { + if err != nil { + text := err.Error() + var se *sidecarError + if errors.As(err, &se) { + text = se.Error() + } + return &mcp.CallToolResult{ + IsError: true, + Content: []mcp.Content{&mcp.TextContent{Text: text}}, + }, nil, nil + } + out, err := json.Marshal(res) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(out)}}, + }, nil, nil +} diff --git a/cmd/sam-a2a-bridge/main_test.go b/cmd/sam-a2a-bridge/main_test.go new file mode 100644 index 00000000..91ebd28a --- /dev/null +++ b/cmd/sam-a2a-bridge/main_test.go @@ -0,0 +1,129 @@ +// 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 main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +func textContent(t *testing.T, res *mcp.CallToolResult) string { + t.Helper() + if len(res.Content) != 1 { + t.Fatalf("content items = %d, want 1", len(res.Content)) + } + tc, ok := res.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("content type = %T, want *mcp.TextContent", res.Content[0]) + } + return tc.Text +} + +func TestHandleSendAgentTaskSuccess(t *testing.T) { + // SendMessage results arrive enveloped as {"task":...} per a2a.StreamResponse (a2a_test.go). + task := `{"task":{"kind":"task","id":"t1","contextId":"c1","status":{"state":"completed",` + + `"message":{"kind":"message","messageId":"m1","role":"agent","parts":[{"kind":"text","text":"done"}]}}}}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", task, nil) + defer srv.Close() + + res, _, err := mcpTool(bridgeConfig{sidecarURL: srv.URL}, handleSendAgentTask)( + context.Background(), nil, sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", Message: "hi"}) + if err != nil { + t.Fatal(err) + } + if res.IsError { + t.Fatalf("unexpected tool error: %s", textContent(t, res)) + } + var got taskResult + if err := json.Unmarshal([]byte(textContent(t, res)), &got); err != nil { + t.Fatalf("output is not the 4-field JSON: %v", err) + } + if got.TaskID != "t1" || got.Text != "done" { + t.Errorf("got %+v", got) + } +} + +func TestHandleSendAgentTaskRefusalIsToolError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Required labels not attested by provider", http.StatusForbidden) + })) + defer srv.Close() + + res, _, err := mcpTool(bridgeConfig{sidecarURL: srv.URL}, handleSendAgentTask)( + context.Background(), nil, sendAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", + Message: "hi", RequiredLabels: "region=us-east-1"}) + if err != nil { + t.Fatalf("refusals are tool errors, not protocol errors: %v", err) + } + if !res.IsError { + t.Fatal("IsError must be true on sidecar refusal") + } + text := textContent(t, res) + if !strings.HasPrefix(text, "403: Required labels not attested by provider") { + t.Errorf("refusal must lead with status + verbatim body, got %q", text) + } +} + +func TestHandleGetAgentTask(t *testing.T) { + task := `{"kind":"task","id":"t9","contextId":"c9","status":{"state":"completed"},` + + `"artifacts":[{"artifactId":"a1","parts":[{"kind":"text","text":"result"}]}]}` + srv := fakeSidecar(t, "/sam/12D3KooWpeer/a2a/echo", task, nil) + defer srv.Close() + + res, _, err := mcpTool(bridgeConfig{sidecarURL: srv.URL}, handleGetAgentTask)( + context.Background(), nil, getAgentTaskParams{Peer: "12D3KooWpeer", Service: "echo", TaskID: "t9"}) + if err != nil { + t.Fatal(err) + } + var got taskResult + if err := json.Unmarshal([]byte(textContent(t, res)), &got); err != nil { + t.Fatal(err) + } + if got.TaskID != "t9" || got.Text != "result" { + t.Errorf("got %+v", got) + } +} + +func TestNewBridgeServerConstructs(t *testing.T) { + if s := newBridgeServer(bridgeConfig{sidecarURL: "http://localhost:8080"}); s == nil { + t.Fatal("newBridgeServer returned nil") + } +} + +func TestHandleGetAgentCard(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"name":"echo-agent","capabilities":{"streaming":true},"skills":[]}`)) + })) + defer srv.Close() + + res, _, err := mcpTool(bridgeConfig{sidecarURL: srv.URL}, handleGetAgentCard)( + context.Background(), nil, getAgentCardParams{Peer: "12D3KooWpeer", Service: "echo"}) + if err != nil { + t.Fatal(err) + } + if res.IsError { + t.Fatalf("unexpected tool error: %s", textContent(t, res)) + } + if !strings.Contains(textContent(t, res), `"echo-agent"`) { + t.Errorf("card summary missing: %s", textContent(t, res)) + } +} diff --git a/cmd/sam-a2a-bridge/transport.go b/cmd/sam-a2a-bridge/transport.go new file mode 100644 index 00000000..410fd86e --- /dev/null +++ b/cmd/sam-a2a-bridge/transport.go @@ -0,0 +1,68 @@ +// 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 main + +import ( + "fmt" + "io" + "net/http" + "strings" +) + +// Mirrors api.HeaderSamAuthentication / api.HeaderSamRequiredLabels in the root module's api/network.go; +// literals because this module must not depend on it. +const ( + headerSamAuthentication = "X-Sam-Authentication" + headerSamRequiredLabels = "X-Sam-Required-Labels" +) + +// sidecarError carries a non-2xx sidecar reply so tool handlers can show +// the refusal (e.g. the labels-gate 403) to the harness verbatim. +type sidecarError struct { + Status int + Body string +} + +func (e *sidecarError) Error() string { + return fmt.Sprintf("%d: %s", e.Status, strings.TrimSpace(e.Body)) +} + +// samTransport injects the local sidecar gate headers on every outbound +// request and converts non-2xx replies into sidecarError. +type samTransport struct { + base http.RoundTripper + token string + requiredLabels string +} + +func (t *samTransport) RoundTrip(req *http.Request) (*http.Response, error) { + req = req.Clone(req.Context()) + if t.token != "" { + req.Header.Set(headerSamAuthentication, "Bearer "+t.token) + } + if t.requiredLabels != "" { + req.Header.Set(headerSamRequiredLabels, t.requiredLabels) + } + resp, err := t.base.RoundTrip(req) + if err != nil { + return nil, err + } + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + _ = resp.Body.Close() + return nil, &sidecarError{Status: resp.StatusCode, Body: string(body)} + } + return resp, nil +} diff --git a/cmd/sam-a2a-bridge/transport_test.go b/cmd/sam-a2a-bridge/transport_test.go new file mode 100644 index 00000000..3fe801cf --- /dev/null +++ b/cmd/sam-a2a-bridge/transport_test.go @@ -0,0 +1,96 @@ +// 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 main + +import ( + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestSamTransportInjectsHeaders(t *testing.T) { + var gotAuth, gotLabels string + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("X-Sam-Authentication") + gotLabels = r.Header.Get("X-Sam-Required-Labels") + _, _ = w.Write([]byte("ok")) + })) + defer backend.Close() + + client := &http.Client{Transport: &samTransport{base: http.DefaultTransport, token: "tok", requiredLabels: "region=eu"}} + resp, err := client.Get(backend.URL) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if string(body) != "ok" { + t.Fatalf("body = %q", body) + } + if gotAuth != "Bearer tok" { + t.Errorf("X-Sam-Authentication = %q, want %q", gotAuth, "Bearer tok") + } + if gotLabels != "region=eu" { + t.Errorf("X-Sam-Required-Labels = %q, want %q", gotLabels, "region=eu") + } +} + +func TestSamTransportOmitsEmptyHeaders(t *testing.T) { + var sawAuth, sawLabels bool + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, sawAuth = r.Header["X-Sam-Authentication"] + _, sawLabels = r.Header["X-Sam-Required-Labels"] + })) + defer backend.Close() + + client := &http.Client{Transport: &samTransport{base: http.DefaultTransport}} + resp, err := client.Get(backend.URL) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if sawAuth || sawLabels { + t.Fatalf("empty config must not send gate headers (auth=%v labels=%v)", sawAuth, sawLabels) + } +} + +func TestSamTransportMapsRefusalToSidecarError(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Required labels not attested by provider", http.StatusForbidden) + })) + defer backend.Close() + + client := &http.Client{Transport: &samTransport{base: http.DefaultTransport, token: "tok"}} + _, err := client.Get(backend.URL) + if err == nil { + t.Fatal("403 must surface as an error") + } + var se *sidecarError + if !errors.As(err, &se) { + t.Fatalf("error %T does not unwrap to *sidecarError: %v", err, err) + } + if se.Status != http.StatusForbidden { + t.Errorf("status = %d, want 403", se.Status) + } + if !strings.Contains(se.Body, "Required labels not attested by provider") { + t.Errorf("body not verbatim: %q", se.Body) + } + if want := "403: Required labels not attested by provider"; se.Error() != want { + t.Errorf("Error() = %q, want %q", se.Error(), want) + } +} diff --git a/development/examples/chat-a2a/agent.py b/development/examples/chat-a2a/agent.py index b3802826..95d0f9db 100644 --- a/development/examples/chat-a2a/agent.py +++ b/development/examples/chat-a2a/agent.py @@ -1,7 +1,9 @@ """Gemini-backed A2A chat agent hosted by a node in the local dev mesh.""" +import json import os import time -import uuid + +from google.protobuf.json_format import MessageToDict import uvicorn from a2a.server.agent_execution.agent_executor import AgentExecutor @@ -10,14 +12,15 @@ 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.server.tasks.task_updater import TaskUpdater +from a2a.helpers.proto_helpers import new_task from a2a.types import ( AgentCapabilities, AgentCard, AgentInterface, AgentSkill, - Message, Part, - Role, + TaskState, ) from google import genai from google.genai import types @@ -26,12 +29,41 @@ PORT = 7777 MODEL = os.environ.get("GEMINI_MODEL", "models/gemini-3.5-flash-lite") +# Typed channel for "I need the user to answer first": a function call is +# schema-enforced, unlike a magic reply prefix the model may forget or misquote. +ASK_USER = types.FunctionDeclaration( + name="ask_user", + description="Ask the user a clarifying question you need answered before you can complete the request.", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={"question": types.Schema(type=types.Type.STRING)}, + required=["question"], + ), +) + +# Same typed channel for producing files: the content becomes an A2A artifact +# (raw bytes + filename) instead of being pasted into the chat reply. +RETURN_FILE = types.FunctionDeclaration( + name="return_file", + description="Deliver a generated text-based file (CSV, Markdown, JSON, plain text) to the user as a downloadable attachment instead of pasting it into the reply.", + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "filename": types.Schema(type=types.Type.STRING), + "content": types.Schema(type=types.Type.STRING), + "media_type": types.Schema(type=types.Type.STRING, description="MIME type, e.g. text/csv"), + }, + required=["filename", "content"], + ), +) + class ChatExecutor(AgentExecutor): """One Gemini chat session per A2A contextId; the session carries the history.""" def __init__(self): self.gemini = genai.Client() self.chats = {} + self.pending_question = set() async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: chat = self.chats.get(context.context_id) @@ -40,26 +72,75 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non chat = self.gemini.aio.chats.create( model=MODEL, config=types.GenerateContentConfig( - thinking_config=types.ThinkingConfig(thinking_level="minimal") + thinking_config=types.ThinkingConfig(thinking_level="minimal"), + tools=[types.Tool(function_declarations=[ASK_USER, RETURN_FILE])], ), ) self.chats[context.context_id] = chat + prompt = context.get_user_input() + data_parts = [ + MessageToDict(part.data) + for part in context.message.parts + if part.WhichOneof("content") == "data" + ] + if data_parts: + # Text-first agent: structured payloads reach Gemini as a labeled block. + prompt += "\n[structured data]: " + json.dumps(data_parts) + gemini_parts = [prompt] + for part in context.message.parts: + if part.WhichOneof("content") != "raw": + continue + media = part.media_type or "application/octet-stream" + if media.startswith("text/"): + filename = part.filename or "unnamed" + # Text attachments read best as prompt text, not opaque blobs. + gemini_parts[0] += f"\n[attached file {filename}]:\n" + part.raw.decode("utf-8", "replace") + else: + gemini_parts.append(types.Part.from_bytes(data=part.raw, mime_type=media)) + # A dangling ask_user call must be answered in-history; the user's + # reply IS the tool response. + if context.context_id in self.pending_question: + self.pending_question.discard(context.context_id) + gemini_parts[0] = types.Part.from_function_response( + name="ask_user", response={"answer": gemini_parts[0]} + ) started = time.monotonic() - reply = await chat.send_message(context.get_user_input()) + reply = await chat.send_message(gemini_parts) 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, + updater = TaskUpdater(event_queue, context.task_id, context.context_id) + # The runtime rejects a status update as a task's first event. + if context.current_task is None: + await event_queue.enqueue_event( + new_task(context.task_id, context.context_id, TaskState.TASK_STATE_SUBMITTED) ) - ) + calls = reply.function_calls or [] + if calls and calls[0].name == "ask_user": + self.pending_question.add(context.context_id) + question = str(calls[0].args.get("question", "")) + await updater.requires_input(updater.new_agent_message([Part(text=question)])) + return + if calls and calls[0].name == "return_file": + filename = str(calls[0].args.get("filename", "file.txt")) + content = str(calls[0].args.get("content", "")) + media = str(calls[0].args.get("media_type") or "text/plain") + await updater.add_artifact( + [Part(raw=content.encode("utf-8"), filename=filename, media_type=media)], + name=filename, + ) + # Unlike ask_user, the tool result is known now: answer the call + # immediately so the chat history stays valid for the next turn. + reply = await chat.send_message( + types.Part.from_function_response(name="return_file", response={"delivered": True}) + ) + await updater.complete( + updater.new_agent_message([Part(text=reply.text or f"Sent {filename}.")]) + ) + return + await updater.complete(updater.new_agent_message([Part(text=reply.text or "")])) async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: pass @@ -70,8 +151,8 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None 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"], + default_input_modes=["text/plain", "application/json", "application/pdf", "image/png", "image/jpeg"], + default_output_modes=["text/plain", "text/csv", "text/markdown", "application/json"], skills=[ AgentSkill( id="chat",