diff --git a/README.md b/README.md index 07cce1e..236836a 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,10 @@ a2a send -a https://agent.example.com "Hello, what can you do?" # Stream events as they arrive a2a send -a https://agent.example.com --stream "Summarize this document" + +# Send asynchronously and poll until completion +a2a send -a https://agent.example.com --async "Analyze this dataset" +a2a task get -a https://agent.example.com --wait --timeout 60s ``` See the **[full command reference](./internal/README.md)** for all commands, flags, configuration, and server mode. diff --git a/agent-plugin/skills/a2a-cli/SKILL.md b/agent-plugin/skills/a2a-cli/SKILL.md index 6cbd91e..f11490c 100644 --- a/agent-plugin/skills/a2a-cli/SKILL.md +++ b/agent-plugin/skills/a2a-cli/SKILL.md @@ -96,10 +96,13 @@ transport) or `-e --transport ` (connect directly). a2a task subscribe -a https://agent.example.com ``` -4. **Check status and fetch results** at any time: +4. **Check status, poll to completion, or fetch results:** ```bash a2a task get -a https://agent.example.com + # Poll until completion or input/auth-required: + a2a task get -a https://agent.example.com --wait + a2a task get -a https://agent.example.com --wait --poll-interval 2s --timeout 60s ``` 5. **Answer an interrupted task** (it reached `INPUT_REQUIRED` / `AUTH_REQUIRED`) @@ -132,6 +135,9 @@ Run `a2a --help` for the full, current set. The load-bearing ones: | `-e, --endpoint ` + `--transport ` | Connect to one interface directly, skipping card resolution. | | `-o, --output json` | Machine-readable output; add `--stream` for a live event stream. | | `--async` | Return immediately with the identifiers instead of blocking; poll later with `task get`. | +| `--wait` | Poll `task get` repeatedly until reaching a terminal or interrupted state. | +| `--poll-interval ` | Delay between polls while waiting (default `2s`). | +| `--timeout ` | Request timeout and overall wait budget (default `30s`). | | `--task-id ` / `--context-id ` | Continue a task / group a new task under a context. | | `--auth ""` / `--svc-param ` | Attach credentials or transport parameters (or set `A2ACLI_*` env vars). Never commit a secret. | diff --git a/examples/04-messages-and-tasks/README.md b/examples/04-messages-and-tasks/README.md index 1359a3b..f47e1dd 100644 --- a/examples/04-messages-and-tasks/README.md +++ b/examples/04-messages-and-tasks/README.md @@ -76,6 +76,8 @@ Once you have a task id, the `task` commands inspect and manage it: ```bash a2a task get # inspect one task +a2a task get --wait # poll until the task finishes +a2a task get --wait --poll-interval 1s --timeout 60s a2a task get --history 10 -o json a2a task list --status completed --limit 20 # list recent tasks a2a task subscribe # follow it until it finishes diff --git a/internal/cli/cli_test.go b/internal/cli/cli_test.go index fb29e2a..19d3d69 100644 --- a/internal/cli/cli_test.go +++ b/internal/cli/cli_test.go @@ -177,6 +177,61 @@ func TestCardGet(t *testing.T) { }) } +func TestCardGet_Timeout(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(100 * time.Millisecond) + })) + t.Cleanup(server.Close) + + _, err := runCMD(t, "card", "get", server.URL, "--timeout", "10ms") + if err == nil { + t.Fatal("a2a card get should time out") + } + var classified *clierr.Error + if !errors.As(err, &classified) { + t.Fatalf("runCMD(card get) error = %T, want *clierr.Error", err) + } + if classified.Code != clierr.CodeTimeout { + t.Fatalf("runCMD(card get) Code = %q, want %q", classified.Code, clierr.CodeTimeout) + } + if classified.Exit != 5 { + t.Fatalf("runCMD(card get) Exit = %d, want 5", classified.Exit) + } + if classified.Hint != "increase --timeout" { + t.Fatalf("runCMD(card get) Hint = %q, want %q", classified.Hint, "increase --timeout") + } +} + +func TestCardGet_TimeoutBoundsHTTPRequests(t *testing.T) { + t.Parallel() + url := startTestServer(t) + + recorder := &deadlineRecorder{} + cfg := &globalConfig{ + Printer: output.NewPrinter(&bytes.Buffer{}, output.ModeText), + svcParams: &flagparse.ServiceParams{}, + httpTransport: recorder, + } + root := mustNewRoot(t, cfg, deps{poller: polling.Stream, cfgLoader: clicfg.LoadEmpty}) + root.SetArgs([]string{"card", "get", url, "--timeout", "25m"}) + + start := time.Now() + if err := root.Execute(); err != nil { + t.Fatalf("root.Execute(card get) error = %v", err) + } + + deadlines := recorder.get() + if len(deadlines) == 0 { + t.Fatal("card get did not issue any request through the configured HTTP transport") + } + for _, d := range deadlines { + if d.IsZero() || d.Before(start.Add(20*time.Minute)) { + t.Fatalf("request deadline = %v after start, want the 25m --timeout", d.Sub(start)) + } + } +} + func TestVersion(t *testing.T) { t.Parallel() @@ -932,8 +987,8 @@ func TestGetTaskWait_Timeout(t *testing.T) { if err == nil { t.Fatal("a2a task get --wait against a never-terminal task should time out") } - if !strings.Contains(err.Error(), "timed out") { - t.Fatalf("a2a task get --wait error = %v, want a timeout error", err) + if got := clierr.Classify(err).Code; got != clierr.CodeTimeout { + t.Fatalf("a2a task get --wait error = %v (code %q), want code %q", err, got, clierr.CodeTimeout) } } @@ -960,6 +1015,32 @@ func TestServe_ModeValidation(t *testing.T) { } } +func TestServeProxy_Timeout(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(100 * time.Millisecond) + })) + t.Cleanup(server.Close) + + _, err := runCMD(t, "server", "--proxy", "-a", server.URL, "--port", "0", "--timeout", "10ms") + if err == nil { + t.Fatal("a2a server --proxy should time out resolving upstream card") + } + var classified *clierr.Error + if !errors.As(err, &classified) { + t.Fatalf("runCMD(server --proxy) error = %v, want *clierr.Error", err) + } + if classified.Code != clierr.CodeTimeout { + t.Fatalf("runCMD(server --proxy) Code = %q, want %q", classified.Code, clierr.CodeTimeout) + } + if classified.Exit != 5 { + t.Fatalf("runCMD(server --proxy) Exit = %d, want 5", classified.Exit) + } + if classified.Hint != "increase --timeout" { + t.Fatalf("runCMD(server --proxy) Hint = %q, want %q", classified.Hint, "increase --timeout") + } +} + func TestConfigApplied(t *testing.T) { t.Parallel() url := startTestServer(t) diff --git a/internal/cli/client.go b/internal/cli/client.go index 0e14a2d..3158a44 100644 --- a/internal/cli/client.go +++ b/internal/cli/client.go @@ -20,7 +20,6 @@ import ( "net/http" "net/url" "strings" - "time" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -36,12 +35,6 @@ import ( a2agrpc "github.com/a2aproject/a2a-go/v2/a2agrpc/v1" ) -var compatCardResolver = func() *agentcard.Resolver { - resolver := agentcard.NewResolver(&http.Client{Timeout: 30 * time.Second}) - resolver.CardParser = a2av0.NewAgentCardParser() - return resolver -}() - const insecureCredentialWarning = "warning: sending a credential with TLS verification disabled (--insecure); the token may be exposed." // warnInsecureCredential warns once when a credential is sent with --insecure. @@ -136,13 +129,19 @@ func resolveCard(ctx context.Context, cfg *globalConfig, ref string) (*a2a.Agent opts = append(opts, agentcard.WithRequestHeader(k, strings.Join(vals, ", "))) } cfg.logf("resolving agent card from %s", ref) - card, err := compatCardResolver.Resolve(ctx, ref, opts...) + card, err := cardResolver(cfg).Resolve(ctx, ref, opts...) if err != nil { return nil, clierr.CardResolution(err) } return card, nil } +func cardResolver(cfg *globalConfig) *agentcard.Resolver { + resolver := agentcard.NewResolver(&http.Client{Transport: cfg.httpTransport}) + resolver.CardParser = a2av0.NewAgentCardParser() + return resolver +} + // hintInsecure wraps gRPC "no transport security set" errors with a user-friendly suggestion to pass --insecure. func hintInsecure(err error) error { if err != nil && strings.Contains(err.Error(), "no transport security set") { diff --git a/internal/cli/serve.go b/internal/cli/serve.go index 81617dc..2feb01a 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -110,7 +110,9 @@ func newServeCmd(cfg *globalConfig) *cobra.Command { if !quiet { proxyOpts = append(proxyOpts, localsrv.ProxyLogInterceptorOption()) } - client, err := newAgentClient(ctx, cfg, proxyOpts...) + clientCtx, clientCancel := context.WithTimeout(ctx, cfg.timeout) + client, err := newAgentClient(clientCtx, cfg, proxyOpts...) + clientCancel() if err != nil { return fmt.Errorf("creating upstream client: %w", err) } diff --git a/internal/clierr/clierr.go b/internal/clierr/clierr.go index 87c9b40..c5f3052 100644 --- a/internal/clierr/clierr.go +++ b/internal/clierr/clierr.go @@ -89,6 +89,15 @@ func Usage(msg string) *Error { // it was reached but did not yield a usable card. func CardResolution(err error) *Error { msg := fmt.Sprintf("failed to resolve agent card: %v", err) + if errors.Is(err, context.DeadlineExceeded) { + return &Error{ + Code: CodeTimeout, + Message: msg, + Exit: 5, + err: err, + Hint: "increase --timeout", + } + } if isIOError(err) { return &Error{ Code: CodeIO, diff --git a/internal/clierr/clierr_test.go b/internal/clierr/clierr_test.go index 6dc9255..aee4cac 100644 --- a/internal/clierr/clierr_test.go +++ b/internal/clierr/clierr_test.go @@ -47,6 +47,12 @@ func TestClassify(t *testing.T) { wantCode: CodeCardInvalid, wantExit: 4, }, + { + name: "card resolution deadline exceeded is a timeout", + err: CardResolution(context.DeadlineExceeded), + wantCode: CodeTimeout, + wantExit: 5, + }, { name: "deadline exceeded is a timeout", err: fmt.Errorf("waiting: %w", context.DeadlineExceeded), @@ -136,18 +142,28 @@ func TestCardResolution(t *testing.T) { err error wantCode string wantExit int + wantHint string }{ { name: "transport failure is io", err: fmt.Errorf("card request failed: %w", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ECONNREFUSED}), wantCode: CodeIO, wantExit: 3, + wantHint: "check the agent is running and the --agent-card reference is correct", }, { name: "non-transport failure is card-invalid", err: errors.New("card request failed, status: 404 Not Found"), wantCode: CodeCardInvalid, wantExit: 4, + wantHint: "check the --agent-card reference (host, full card URL, or local file path)", + }, + { + name: "deadline exceeded is a timeout", + err: context.DeadlineExceeded, + wantCode: CodeTimeout, + wantExit: 5, + wantHint: "increase --timeout", }, } @@ -156,13 +172,16 @@ func TestCardResolution(t *testing.T) { t.Parallel() got := CardResolution(tt.err) if got.Code != tt.wantCode { - t.Errorf("CardResolution(%v).Code = %q, want %q", tt.err, got.Code, tt.wantCode) + t.Fatalf("CardResolution(%v).Code = %q, want %q", tt.err, got.Code, tt.wantCode) } if got.Exit != tt.wantExit { - t.Errorf("CardResolution(%v).Exit = %d, want %d", tt.err, got.Exit, tt.wantExit) + t.Fatalf("CardResolution(%v).Exit = %d, want %d", tt.err, got.Exit, tt.wantExit) + } + if got.Hint != tt.wantHint { + t.Fatalf("CardResolution(%v).Hint = %q, want %q", tt.err, got.Hint, tt.wantHint) } if !errors.Is(got, tt.err) { - t.Errorf("CardResolution(%v) does not wrap the cause", tt.err) + t.Fatalf("CardResolution(%v) does not wrap the cause", tt.err) } }) } diff --git a/skills/a2a-cli/SKILL.md b/skills/a2a-cli/SKILL.md index d99e154..f05498d 100644 --- a/skills/a2a-cli/SKILL.md +++ b/skills/a2a-cli/SKILL.md @@ -96,10 +96,13 @@ transport) or `-e --transport ` (connect directly). a2a task subscribe -a https://agent.example.com ``` -4. **Check status and fetch results** at any time: +4. **Check status, poll to completion, or fetch results:** ```bash a2a task get -a https://agent.example.com + # Poll until completion or input/auth-required: + a2a task get -a https://agent.example.com --wait + a2a task get -a https://agent.example.com --wait --poll-interval 2s --timeout 60s ``` 5. **Answer an interrupted task** (it reached `INPUT_REQUIRED` / `AUTH_REQUIRED`) @@ -132,6 +135,9 @@ Run `a2a --help` for the full, current set. The load-bearing ones: | `-e, --endpoint ` + `--transport ` | Connect to one interface directly, skipping card resolution. | | `-o, --output json` | Machine-readable output; add `--stream` for a live event stream. | | `--async` | Return immediately with the identifiers instead of blocking; poll later with `task get`. | +| `--wait` | Poll `task get` repeatedly until reaching a terminal or interrupted state. | +| `--poll-interval ` | Delay between polls while waiting (default `2s`). | +| `--timeout ` | Request timeout and overall wait budget (default `30s`). | | `--task-id ` / `--context-id ` | Continue a task / group a new task under a context. | | `--save-fileparts ` | Save raw file parts from responses (artifacts and messages) as files under ``; the output still summarizes them by name/type/size. | | `--auth ""` / `--svc-param ` | Attach credentials or transport parameters (or set `A2ACLI_*` env vars). Never commit a secret. |