diff --git a/internal/cli/cli_test.go b/internal/cli/cli_test.go index cee4466..1436b60 100644 --- a/internal/cli/cli_test.go +++ b/internal/cli/cli_test.go @@ -241,11 +241,8 @@ func TestSend(t *testing.T) { if err != nil { t.Fatalf("runCMD(%q) error = %v", strings.Join(tt.args(mode.url), " "), err) } - var task a2a.Task - if err := json.Unmarshal([]byte(out), &task); err != nil { - t.Fatalf("json.Unmarshal() error = %v", err) - } - if text := testutil.AllArtifactText(&task); text != tt.wantText { + task := mustDecodeTask(t, out) + if text := testutil.AllArtifactText(task); text != tt.wantText { t.Fatalf("allArtifactText() = %q, want %q", text, tt.wantText) } }) @@ -285,11 +282,8 @@ func TestSend_AgentCardFromFile(t *testing.T) { if err != nil { t.Fatalf("runCMD(%q) error = %v", strings.Join(tt.args, " "), err) } - var task a2a.Task - if err := json.Unmarshal([]byte(out), &task); err != nil { - t.Fatalf("json.Unmarshal() error = %v", err) - } - if text := testutil.AllArtifactText(&task); text != tt.wantText { + task := mustDecodeTask(t, out) + if text := testutil.AllArtifactText(task); text != tt.wantText { t.Fatalf("allArtifactText() = %q, want %q", text, tt.wantText) } }) @@ -306,11 +300,8 @@ func TestSendDataPart(t *testing.T) { } out := mustRunCMD(t, "send", "-a", url, "-o", "json", "--data-part", path) - var task a2a.Task - if err := json.Unmarshal([]byte(out), &task); err != nil { - t.Fatalf("json.Unmarshal(send --data-part output) error = %v", err) - } - if got := testutil.AllArtifactText(&task); got != `{"hello":"world"}` { + task := mustDecodeTask(t, out) + if got := testutil.AllArtifactText(task); got != `{"hello":"world"}` { t.Fatalf("allArtifactText() = %q, want %q", got, `{"hello":"world"}`) } } @@ -325,11 +316,8 @@ func TestSendRequestPayloadFile(t *testing.T) { } out := mustRunCMD(t, "send", "-a", url, "-o", "json", "--request-payload", path) - var task a2a.Task - if err := json.Unmarshal([]byte(out), &task); err != nil { - t.Fatalf("json.Unmarshal(send --request-payload output) error = %v", err) - } - if got := testutil.AllArtifactText(&task); got != "from file" { + task := mustDecodeTask(t, out) + if got := testutil.AllArtifactText(task); got != "from file" { t.Fatalf("allArtifactText() = %q, want %q", got, "from file") } } @@ -490,12 +478,61 @@ func TestSendStreaming(t *testing.T) { } } +func TestSendStreamJSONL(t *testing.T) { + t.Parallel() + url := startTestServer(t) + + testCases := []struct { + name string + flags []string + wantObjectPerLine bool + }{ + { + name: "compact one object per line", + flags: []string{"-a", url, "-o", "json", "--stream"}, + wantObjectPerLine: true, + }, + { + name: "indented objects when pretty", + flags: []string{"-a", url, "-o", "json", "--stream", "--pretty"}, + wantObjectPerLine: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + command := append([]string{"send", "stream me"}, tc.flags...) + out := mustRunCMD(t, command...) + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + if len(lines) == 0 { + t.Fatalf("send --stream produced no JSONL lines") + } + objectPerLine := true + for i, line := range lines { + var sr a2a.StreamResponse + if err := json.Unmarshal([]byte(line), &sr); err != nil { + if tc.wantObjectPerLine { + t.Fatalf("JSONL line %d is not an independently parseable object: %v\nline: %s", i, err, line) + } + objectPerLine = false + break + } + } + if objectPerLine && !tc.wantObjectPerLine { + t.Fatalf("all outputs lines contained a well-formed a2a.StreamResponse:\n%s", out) + } + }) + } + +} + func TestSendStreamingFallbackUsesDefaultPoller(t *testing.T) { t.Parallel() nonStreamingURL := startTestServerWith(t, a2a.AgentCapabilities{Streaming: false}) out, err := runCMDWithConfig(t, deps{cfgLoader: clicfg.LoadEmpty}, - "send", "-a", nonStreamingURL, "-o", "json", "--stream", "stream me", "--polling-interval", "5ms") + "send", "-a", nonStreamingURL, "-o", "json", "--stream", "stream me", "--poll-interval", "5ms") if err != nil { t.Fatalf("runCMDWithConfig() error = %v", err) } @@ -514,6 +551,108 @@ func TestSendStreamingFallbackUsesDefaultPoller(t *testing.T) { } } +func TestSend_ResumeHintForInputRequiredTask(t *testing.T) { + t.Parallel() + + var taskID a2a.TaskID + server := httptest.NewServer(a2asrv.NewRESTHandler(a2asrv.NewHandler( + a2asrv.AgentExecutorFunc(func(ctx context.Context, ec *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] { + return func(yield func(a2a.Event, error) bool) { + taskID = ec.TaskID + task := &a2a.Task{ + ID: ec.TaskID, + ContextID: ec.ContextID, + Status: a2a.TaskStatus{State: a2a.TaskStateInputRequired}, + } + yield(task, nil) + } + }), + ))) + t.Cleanup(server.Close) + + out := mustRunCMD(t, "send", "-e", server.URL, "--transport", "rest", "hello") + if !strings.Contains(out, "a2a send --task-id "+string(taskID)) { + t.Fatalf("send text output missing the resume hint:\n%s", out) + } +} + +func TestSendWithVersionSelector(t *testing.T) { + t.Parallel() + url := startTestServer(t) + legacyURL := startLegacyTestServer(t) + + testCases := []struct { + name string + connect []string + version string + wantErr bool + }{ + { + name: "new server success", + connect: []string{"-a", url}, + version: "1.0", + }, + { + name: "old server success", + connect: []string{"-a", legacyURL}, + version: "0.3", + }, + { + name: "new server direct success", + connect: []string{"-e", url, "--transport", "rest"}, + version: "1.0", + }, + { + name: "old server direct success", + connect: []string{"-e", legacyURL, "--transport", "jsonrpc"}, + version: "0.3", + }, + { + name: "new server failure", + connect: []string{"-a", url}, + version: "0.3", + wantErr: true, + }, + { + name: "new server direct failure", + connect: []string{"-e", url, "--transport", "rest"}, + version: "0.3", + wantErr: true, + }, + { + name: "old server failure", + connect: []string{"-a", legacyURL}, + version: "1.0", + wantErr: true, + }, + { + name: "old server direct failure", + connect: []string{"-e", legacyURL, "--transport", "jsonrpc"}, + version: "1.0", + wantErr: true, + }, + { + name: "unknown version failure", + connect: []string{"-e", url}, + version: "3.0", + wantErr: true, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + command := []string{"send", "--a2a-version", tc.version, "-o", "json", "hi"} + command = append(command, tc.connect...) + _, err := runCMD(t, command...) + if err != nil && !tc.wantErr { + t.Fatalf("send error = %v", err) + } + if err == nil && tc.wantErr { + t.Fatal("send error = nil, wanted a failure") + } + }) + } +} + func TestGetTask(t *testing.T) { t.Parallel() url := startTestServer(t) @@ -695,6 +834,19 @@ func startLegacyTestServer(t *testing.T) string { return server.URL } +func mustDecodeTask(t *testing.T, out string) *a2a.Task { + t.Helper() + var resp a2a.StreamResponse + if err := json.Unmarshal([]byte(out), &resp); err != nil { + t.Fatalf("json.Unmarshal() error = %v\noutput: %s", err, out) + } + task, ok := resp.Event.(*a2a.Task) + if !ok { + t.Fatalf("send output has no task wrapper: %s", out) + } + return task +} + func sendTestMessage(t *testing.T, url, text string) a2a.TaskID { t.Helper() ctx := t.Context() diff --git a/internal/cli/client.go b/internal/cli/client.go index 5de692c..834a2a4 100644 --- a/internal/cli/client.go +++ b/internal/cli/client.go @@ -67,6 +67,9 @@ func newClientFromEndpoint(ctx context.Context, cfg *globalConfig, ref string, e cfg.logf("connecting directly to %s via %s (skipping card resolution)", endpointURL, protocol) endpoint := a2a.NewAgentInterface(endpointURL, protocol) + if cfg.a2aVersion != "" { + endpoint.ProtocolVersion = a2a.ProtocolVersion(cfg.a2aVersion) + } client, err := a2aclient.NewFromEndpoints(ctx, []*a2a.AgentInterface{endpoint}, append(clientFactoryOpts(cfg), extraOpts...)...) return client, hintInsecure(err) } @@ -108,19 +111,28 @@ func hintInsecure(err error) error { } func clientFactoryOpts(cfg *globalConfig) []a2aclient.FactoryOption { - factoryOpts := []a2aclient.FactoryOption{ - a2av0.WithRESTTransport(a2av0.RESTTransportConfig{}), - a2av0.WithJSONRPCTransport(a2av0.JSONRPCTransportConfig{}), - } var grpcOpts []grpc.DialOption if cfg.insecureGRPC { grpcOpts = append(grpcOpts, grpc.WithTransportCredentials(insecure.NewCredentials())) } - factoryOpts = append(factoryOpts, - a2agrpcv0.WithGRPCTransport(grpcOpts...), - a2agrpc.WithGRPCTransport(grpcOpts...), - ) - return factoryOpts + opts := []a2aclient.FactoryOption{a2aclient.WithDefaultsDisabled()} + if cfg.a2aVersion == "" || cfg.a2aVersion == "1.0" { + opts = append( + opts, + a2aclient.WithRESTTransport(nil), + a2aclient.WithJSONRPCTransport(nil), + a2agrpc.WithGRPCTransport(grpcOpts...), + ) + } + if cfg.a2aVersion == "" || cfg.a2aVersion == "0.3" { + opts = append( + opts, + a2av0.WithRESTTransport(a2av0.RESTTransportConfig{}), + a2av0.WithJSONRPCTransport(a2av0.JSONRPCTransportConfig{}), + a2agrpcv0.WithGRPCTransport(grpcOpts...), + ) + } + return opts } func stripHTTPScheme(raw string) string { diff --git a/internal/cli/root.go b/internal/cli/root.go index 196d3f8..033d79f 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -50,10 +50,12 @@ type globalConfig struct { url string transports []string svcParams *flagparse.ServiceParams + a2aVersion string tenant string timeout time.Duration verbose bool insecureGRPC bool + pretty bool configPath string bindings []clicfg.FlagBinding @@ -107,6 +109,7 @@ func newRootCmd(cfg *globalConfig, deps deps) *cobra.Command { default: return fmt.Errorf("invalid --output %q (want text or json)", cfg.output) } + cfg.Printer.PrettyJSONL = cfg.pretty return nil }, } @@ -116,11 +119,13 @@ func newRootCmd(cfg *globalConfig, deps deps) *cobra.Command { pf.VarP(&cfg.agentCard, "agent-card", "a", "Agent Card reference: host/origin, full card URL, or local file path") pf.StringVarP(&cfg.url, "endpoint", "e", "", "Agent interface URL for a direct connection; skips card resolution and requires a single --transport flag") pf.StringArrayVar(&cfg.transports, "transport", nil, "Transport preference: rest, jsonrpc, grpc (repeatable, highest preference first)") + pf.StringVar(&cfg.a2aVersion, "a2a-version", "", "Controls which a2a-protocol version client will advertise to the server.") cfg.svcParams.Attach(pf) pf.StringVar(&cfg.tenant, "tenant", "", "Tenant identifier") pf.DurationVar(&cfg.timeout, "timeout", 30*time.Second, "Request timeout") pf.BoolVarP(&cfg.verbose, "verbose", "v", false, "Verbose output to stderr") pf.BoolVar(&cfg.insecureGRPC, "insecure", false, "Use insecure (plaintext) gRPC transport credentials") + pf.BoolVar(&cfg.pretty, "pretty", false, "Pretty-print (indent) streamed JSONL records instead of emitting one compact object per line") pf.StringVar(&cfg.configPath, "config", "", "Load configuration from an explicit .env file in place of the local .env") cmd.AddCommand( diff --git a/internal/cli/send.go b/internal/cli/send.go index 2c0b557..0132a67 100644 --- a/internal/cli/send.go +++ b/internal/cli/send.go @@ -31,15 +31,15 @@ import ( ) type sendFlags struct { - stream bool - async bool - payload string - taskID string - contextID string - history int - pollingInterval time.Duration - parts flagparse.Parts - meta flagparse.Metadata + stream bool + async bool + payload string + taskID string + contextID string + history int + pollInterval time.Duration + parts flagparse.Parts + meta flagparse.Metadata } type pollerFunc func(ctx context.Context, client *a2aclient.Client, req *a2a.SendMessageRequest, interval time.Duration) iter.Seq2[a2a.Event, error] @@ -81,9 +81,9 @@ func newSendCmd(cfg *globalConfig, poller pollerFunc) *cobra.Command { return utils.UnpackCause(ctx, err) } - cfg.logf("falling back to polling (%v): %v", flags.pollingInterval, err) + cfg.logf("falling back to polling (%v): %v", flags.pollInterval, err) - for event, err := range poller(ctx, client, req, flags.pollingInterval) { + for event, err := range poller(ctx, client, req, flags.pollInterval) { debounceTimeout() if err := handleStreamEntry(cfg, event, err); err != nil { return utils.UnpackCause(ctx, err) @@ -113,7 +113,7 @@ func newSendCmd(cfg *globalConfig, poller pollerFunc) *cobra.Command { f.StringVar(&flags.taskID, "task-id", "", "Task ID to continue an existing task") f.StringVar(&flags.contextID, "context-id", "", "Context ID to group this turn under") f.IntVar(&flags.history, "history", 0, "Request n history messages in the response") - f.DurationVar(&flags.pollingInterval, "polling-interval", 5*time.Second, "Duration between GetTask requests in polling fallback mode.") + f.DurationVar(&flags.pollInterval, "poll-interval", 2*time.Second, "Duration between GetTask requests in polling fallback mode.") flags.parts.Attach(f) flags.meta.Attach(f, "metadata", "Attach request metadata as a JSON object (repeatable)") diff --git a/internal/flagparse/svcparams.go b/internal/flagparse/svcparams.go index 1426445..add874c 100644 --- a/internal/flagparse/svcparams.go +++ b/internal/flagparse/svcparams.go @@ -16,7 +16,6 @@ package flagparse import ( "fmt" - "strings" "github.com/spf13/pflag" @@ -70,9 +69,9 @@ func (s *ServiceParams) Auth() string { type svcParamValue struct{ s *ServiceParams } func (v *svcParamValue) Set(kv string) error { - k, val, ok := strings.Cut(kv, "=") + k, val, ok := cutServiceParam(kv) if !ok { - return fmt.Errorf("expected key=value, got %q", kv) + return fmt.Errorf("expected key=value or key:value, got %q", kv) } if k == "" { return fmt.Errorf("empty key in %q", kv) @@ -81,6 +80,21 @@ func (v *svcParamValue) Set(kv string) error { return nil } +// cutServiceParam splits a --svc-param argument on whichever of ':' or '=' comes first. +func cutServiceParam(kv string) (key, value string, ok bool) { + sep := -1 + for i := 0; i < len(kv); i++ { + if kv[i] == ':' || kv[i] == '=' { + sep = i + break + } + } + if sep < 0 { + return "", "", false + } + return kv[:sep], kv[sep+1:], true +} + func (v *svcParamValue) String() string { return "" } func (v *svcParamValue) Type() string { return "key=value" } diff --git a/internal/flagparse/svcparams_test.go b/internal/flagparse/svcparams_test.go index b5a88ba..b5871b5 100644 --- a/internal/flagparse/svcparams_test.go +++ b/internal/flagparse/svcparams_test.go @@ -43,6 +43,31 @@ func TestServiceParamsParse(t *testing.T) { args: []string{"--svc-param", "k=a=b"}, want: a2aclient.ServiceParams{"k": {"a=b"}}, }, + { + name: "colon separator", + args: []string{"--svc-param", "x-trace:abc"}, + want: a2aclient.ServiceParams{"x-trace": {"abc"}}, + }, + { + name: "colon value may contain colon", + args: []string{"--svc-param", "redirect:http://example.com"}, + want: a2aclient.ServiceParams{"redirect": {"http://example.com"}}, + }, + { + name: "equals before colon splits on equals", + args: []string{"--svc-param", "url=http://example.com"}, + want: a2aclient.ServiceParams{"url": {"http://example.com"}}, + }, + { + name: "colon before equals splits on colon", + args: []string{"--svc-param", "x-trace:a=b"}, + want: a2aclient.ServiceParams{"x-trace": {"a=b"}}, + }, + { + name: "empty key with colon is an error", + args: []string{"--svc-param", ":value"}, + wantErr: true, + }, { name: "repeated keys append in order", args: []string{"--svc-param", "k=1", "--svc-param", "k=2"}, diff --git a/internal/output/output.go b/internal/output/output.go index 14bf4b7..bfba40b 100644 --- a/internal/output/output.go +++ b/internal/output/output.go @@ -50,6 +50,8 @@ var taskStateNames = map[a2a.TaskState]string{ type Printer struct { Out io.Writer Mode Mode + // PrettyJSONL enable JSON indentation for jsonl printing. + PrettyJSONL bool } // NewPrinter returns a Printer that writes to out using the given Mode. @@ -57,7 +59,7 @@ func NewPrinter(out io.Writer, mode Mode) *Printer { return &Printer{Out: out, Mode: mode} } -// PrintJSON writes v as indented JSON. +// PrintJSON writes v as an indented, single JSON document. func (p *Printer) PrintJSON(v any) error { enc := json.NewEncoder(p.Out) enc.SetIndent("", " ") @@ -78,15 +80,22 @@ func (p *Printer) PrintTask(task *a2a.Task) error { if p.Mode == ModeJson { return p.PrintJSON(task) } - _, err := io.WriteString(p.Out, formatTask(task)) + _, err := io.WriteString(p.Out, formatTask(task)+formatResumeHint(task)) return err } -// PrintEvent writes a streaming event in the configured Mode. +// PrintEvent writes a streaming event in the configured Mode. In json mode each +// event is emitted as one JSONL record. func (p *Printer) PrintEvent(event a2a.Event) error { if p.Mode == ModeJson { - return p.PrintJSON(a2a.StreamResponse{Event: event}) + enc := json.NewEncoder(p.Out) + enc.SetEscapeHTML(false) + if p.PrettyJSONL { + enc.SetIndent("", " ") + } + return enc.Encode(a2a.StreamResponse{Event: event}) } + var s string switch e := event.(type) { case *a2a.TaskStatusUpdateEvent: @@ -117,17 +126,18 @@ func (p *Printer) PrintEvent(event a2a.Event) error { // PrintSendResult writes the result of a send-message call in the configured Mode. func (p *Printer) PrintSendResult(result a2a.SendMessageResult) error { if p.Mode == ModeJson { - return p.PrintJSON(result) + return p.PrintJSON(a2a.StreamResponse{Event: result}) } + switch r := result.(type) { case *a2a.Task: - _, err := io.WriteString(p.Out, formatTask(r)) + _, err := io.WriteString(p.Out, formatTask(r)+formatResumeHint(r)) return err case *a2a.Message: _, err := io.WriteString(p.Out, formatMessage(r)) return err } - return nil + return fmt.Errorf("unexpected send result type %T", result) } // PrintTaskList writes a list of tasks in the configured Mode. @@ -263,6 +273,17 @@ func formatTask(task *a2a.Task) string { return sb.String() } +// formatResumeHint returns a copy-pasteable command to continue or reply to a task. +func formatResumeHint(task *a2a.Task) string { + if task.ID == "" { + return "" + } + if task.Status.State != a2a.TaskStateInputRequired && task.Status.State != a2a.TaskStateAuthRequired { + return "" + } + return fmt.Sprintf("\n\nResume: a2a send --task-id %s %q\n", task.ID, "") +} + func formatMessage(msg *a2a.Message) string { role := "user" if msg.Role == a2a.MessageRoleAgent { diff --git a/specification/SPEC.md b/specification/SPEC.md index f93bc8f..c896a91 100644 --- a/specification/SPEC.md +++ b/specification/SPEC.md @@ -168,6 +168,7 @@ Task-status **polling** is not a separate command: it is `task get --wait` — r | `--metadata ` | Attach caller-supplied metadata to the message/request as an inline JSON object string (e.g. `'{"k":"v"}'`); values may be any JSON. Sent in the request **payload** (A2A §3.2.5); distinct from `--svc-param`, which sets transport-level parameters (A2A §3.2.6). | | `-o, --output ` | Output **format** only: `text` (default, §6.5, §11.2) or `json`, the protocol's own types (Appendix B). Whether `json` is one document or JSONL is set by `--stream`, not this flag (§11.3). | | `--poll-interval ` / `--timeout ` | How often to re-check task status while waiting, and how long to wait before giving up (§9.3). | +| `--pretty` | Pretty-print (indent) `-o json --stream` records for human reading instead of the default one-compact-object-per-line JSONL (§11.3). Affects presentation only, not the data. OPTIONAL. | | `--stream` | Follow the agent's live event stream instead of blocking, on `send` and `task subscribe`. Sets output **delivery** (peer of `-o`); with `-o json`, emits JSONL (§11.3). Explicit-only. Falls back to polling if the server does not support streaming. | | `--svc-param ` | Add an A2A **service parameter** (A2A §3.2.6): a transport-level key-value pair the binding carries in its own mechanism — an HTTP header or gRPC metadata — repeatable; general-purpose, not authentication-specific (§12.1). Keys and values are strings. Distinct from `--metadata`, which travels in the request payload (A2A §3.2.5). | | `--task-id ` | Continue a specific existing task — for example, to reply to one waiting in `INPUT_REQUIRED`. `--context-id` is optional (the server resolves the task's context) but MUST correspond when given; a rejected identifier fails rather than starting a new task (§8.1). | @@ -414,6 +415,7 @@ The machine-readable format is `-o json`. Its **cardinality follows `--stream`** - **Without `--stream`, `json` MUST be a single document** — exactly one object, never a concatenation of events, so a consumer can `JSON.parse` stdout in one shot. Size is bounded by the task, not by how many events it produced. - **With `--stream`, `json` MUST be emitted as JSONL** — each line a complete, independently parseable JSON object terminated by a newline, flushed as produced. Lines MUST NOT be pretty-printed across multiple physical lines, and the final line MUST carry the terminal event (Appendix B), so a reader that keeps only the last line still obtains the identifiers and state. +- A tool MAY offer `--pretty` as an explicit, human-facing opt-out that indents each streamed record for reading; the default without `--pretty` MUST remain one compact object per physical line. A machine consumer MUST NOT pass `--pretty` (indented records are no longer one-object-per-line), and a tool MUST NOT enable it implicitly. - The switch is **caller-controlled, not implicit**: `--stream` MUST be an explicit command-line flag, and a tool MUST NOT enable it from configuration, an environment variable, or terminal detection. A caller that did not pass `--stream` always gets a single document, so the output shape is a function of the invocation the caller wrote. - If the caller passes `--stream` but the interaction cannot stream — the agent does not advertise the streaming capability, or it returns a `Message` rather than a streamable `Task` — the tool MUST still honor JSONL by emitting the applicable object(s), one per line; a single-line result is valid JSONL. - Both forms MUST emit the A2A protocol's own response types as their **result** (Appendix B); a tool MUST NOT define a substitute result schema. A *failure* instead emits the Appendix B error envelope — the one shape this specification defines of its own (§11.4). @@ -492,7 +494,7 @@ A client MAY express its own preference with `--transport`, which is **repeatabl Cross-cutting options such as `--insecure` apply to whichever transport is negotiated. Where a future option is meaningful only for one binding (for example a gRPC keepalive setting that HTTP has no analogue for), a tool SHOULD namespace it per transport rather than overloading a global flag; the reserved convention is `---