Testing A2A Agents with tamesu: Tasks, Delegations and Agents You Cannot Trust

// The Mochi Shoes concierge: an A2A agent built with Genkit Go. // It answers customers and delegates to two other A2A agents: shipping and billing. package main import ( "context" "fmt" "iter" "log" "net/http" "os" "os/signal" "strings" "sync" "syscall" "time" "github.com/a2aproject/a2a-go/v2/a2a" "github.com/a2aproject/a2a-go/v2/a2aclient" "github.com/a2aproject/a2a-go/v2/a2aclient/agentcard" "github.com/a2aproject/a2a-go/v2/a2acompat/a2av0" "github.com/a2aproject/a2a-go/v2/a2asrv" "github.com/firebase/genkit/go/ai" "github.com/firebase/genkit/go/genkit" "github.com/firebase/genkit/go/plugins/googlegenai" ) // answer is what the model returns on every turn. type answer struct { Reply string `json:"reply" jsonschema:"description=What the concierge says to the customer"` NeedsInput bool `json:"needs_input" jsonschema:"description=True when the reply asks the customer for something the concierge needs to go on"` } type concierge struct { g *genkit.Genkit tools []ai.ToolRef system string mu sync.Mutex sessions map[string][]*ai.Message // one conversation per A2A context } // Execute runs one turn: it is called for every message a client sends. func (c *concierge) Execute(ctx context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] { return func(yield func(a2a.Event, error) bool) { if ec.StoredTask == nil { // a new task if !yield(a2a.NewSubmittedTask(ec, ec.Message), nil) { return } } if !yield(a2a.NewStatusUpdateEvent(ec, a2a.TaskStateWorking, nil), nil) { return } out, err := c.answer(ctx, ec.ContextID, text(ec.Message)) if err != nil { log.Printf("turn failed: %v", err) sorry := a2a.NewMessageForTask(a2a.MessageRoleAgent, ec, a2a.NewTextPart("Sorry, something went wrong on our side.")) yield(a2a.NewStatusUpdateEvent(ec, a2a.TaskStateFailed, sorry), nil) return } if out.NeedsInput { // the task stays open and waits for the customer ask := a2a.NewMessageForTask(a2a.MessageRoleAgent, ec, a2a.NewTextPart(out.Reply)) yield(a2a.NewStatusUpdateEvent(ec, a2a.TaskStateInputRequired, ask), nil) return } if !yield(a2a.NewArtifactEvent(ec, a2a.NewTextPart(out.Reply)), nil) { return } yield(a2a.NewStatusUpdateEvent(ec, a2a.TaskStateCompleted, nil), nil) } } func (c *concierge) Cancel(_ context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] { return func(yield func(a2a.Event, error) bool) { yield(a2a.NewStatusUpdateEvent(ec, a2a.TaskStateCanceled, nil), nil) } } // answer asks Gemini for the next turn, with the history of this context. func (c *concierge) answer(ctx context.Context, contextID, userText string) (*answer, error) { ctx, cancel := context.WithTimeout(ctx, 45*time.Second) // a model call that hangs must not hang the task defer cancel() c.mu.Lock() history := c.sessions[contextID] c.mu.Unlock() out, resp, err := genkit.GenerateData[answer](ctx, c.g, ai.WithSystem(c.system), ai.WithMessages(history...), ai.WithPrompt(userText), ai.WithTools(c.tools...), ai.WithMaxTurns(6), ai.WithCustomConstrainedOutput(), // ask for the JSON in the prompt: with native JSON mode and tools, Gemini keeps calling tools ) if err != nil { return nil, err } var next []*ai.Message for _, m := range resp.History() { if m.Role != ai.RoleSystem { next = append(next, m) } } c.mu.Lock() c.sessions[contextID] = next c.mu.Unlock() return out, nil } // resolver reads agent cards in the 1.0 format and in the 0.3 format. var resolver = &agentcard.Resolver{ Client: &http.Client{Timeout: 10 * time.Second}, CardParser: a2av0.NewAgentCardParser(), } // delegate sends one message to another A2A agent and returns its answer. func delegate(ctx context.Context, baseURL, message string) (string, error) { card, err := resolver.Resolve(ctx, baseURL) if err != nil { return "", fmt.Errorf("agent card: %w", err) } client, err := a2aclient.NewFromCard(ctx, card, a2av0.WithJSONRPCTransport(a2av0.JSONRPCTransportConfig{}), // also talk to agents on protocol 0.3 ) if err != nil { return "", err } defer client.Destroy() msg := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart(message)) res, err := client.SendMessage(ctx, &a2a.SendMessageRequest{Message: msg}) if err != nil { return "", err } switch r := res.(type) { case *a2a.Message: return text(r), nil case *a2a.Task: if r.Status.State == a2a.TaskStateFailed || r.Status.State == a2a.TaskStateRejected { return "", fmt.Errorf("the agent could not do it (task %s)", r.Status.State) } var b strings.Builder b.WriteString(text(r.Status.Message)) for _, art := range r.Artifacts { for _, p := range art.Parts { b.WriteString(p.Text()) } } return b.String(), nil } return "", fmt.Errorf("unexpected result %T", res) } func text(m *a2a.Message) string { if m == nil { return "" } var b strings.Builder for _, p := range m.Parts { b.WriteString(p.Text()) } return b.String() } func main() { ctx := context.Background() g := genkit.Init(ctx, genkit.WithPlugins(&googlegenai.GoogleAI{}), genkit.WithDefaultModel("googleai/gemini-3.8-flash"), ) type ask struct { Message string `json:"message" jsonschema:"description=The full request in plain English, with the order id"` } remote := func(name, description, url string) ai.ToolRef { return genkit.DefineTool(g, name, description, func(tc *ai.ToolContext, in ask) (string, error) { log.Printf("%s <- %q", name, in.Message) reply, err := delegate(tc, url, in.Message) if err != nil { // the model sees the failure and decides what to tell the customer log.Printf("%s failed: %v", name, err) return "ERROR: the agent is not available right now: " + err.Error(), nil } log.Printf("%s -> %q", name, reply) return reply, nil }) } prompt, err := os.ReadFile(envOr("AGENT_PROMPT", "prompt.md")) if err != nil { log.Fatal(err) } agent := &concierge{ g: g, system: string(prompt), sessions: map[string][]*ai.Message{}, tools: []ai.ToolRef{ remote("ask_shipping", "Ask the shipping agent where an order is and when it was or will be delivered.", envOr("SHIPPING_AGENT_URL", "http://127.0.0.1:8096")), remote("ask_billing", "Ask the billing agent to refund an order or to look up an invoice.", envOr("BILLING_AGENT_URL", "http://127.0.0.1:8097")), }, } addr := envOr("ADDR", "127.0.0.1:8100") card := &a2a.AgentCard{ Name: "Mochi Shoes concierge", Description: "Order tracking, returns and refunds for the Mochi Shoes store.", Version: "1.0.0", // Two interfaces: protocol 1.0 at /a2a and protocol 0.3 at /a2a/v0. SupportedInterfaces: []*a2a.AgentInterface{ a2a.NewAgentInterface("http://"+addr+"/a2a", a2a.TransportProtocolJSONRPC), {URL: "http://" + addr + "/a2a/v0", ProtocolBinding: a2a.TransportProtocolJSONRPC, ProtocolVersion: a2av0.Version}, }, DefaultInputModes: []string{"text/plain"}, DefaultOutputModes: []string{"text/plain"}, Capabilities: a2a.AgentCapabilities{Streaming: true}, Skills: []a2a.AgentSkill{ {ID: "order-status", Name: "Order status", Description: "Where an order is and when it arrives", Tags: []string{"orders"}, Examples: []string{"Where is order ORD-1001?"}}, {ID: "returns", Name: "Returns and refunds", Description: "The returns policy, and refunds inside the return window", Tags: []string{"returns"}, Examples: []string{"I want to return order ORD-1001"}}, }, } handler := a2asrv.NewHandler(agent) mux := http.NewServeMux() mux.Handle("/a2a", a2asrv.NewJSONRPCHandler(handler)) mux.Handle("/a2a/v0", a2av0.NewJSONRPCHandler(handler)) mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewAgentCardHandler(a2av0.NewStaticAgentCardProducer(card))) // Genkit installs its own signal handler, so stop the server ourselves on Ctrl-C. srv := &http.Server{Addr: addr, Handler: mux} sig, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop(...