feat(minimal_pvc_agent): support A2A streaming - #1
Open
moonlight16 wants to merge 11 commits into
Open
Conversation
Add a new A2A agent for general-purpose chat using AG2 and llm-d. - Adapted from simple_generalist: single-turn ConversableAgent, no MCP - System prompt: concise helpful-assistant style - Wired for llm-d in-cluster inference gateway (path-prefix routing) - Defaults to meta-llama/Llama-3.3-70B-Instruct - Includes .env templates for local port-forward testing - Added llm-d + llm-d-small blocks to sample-environments.yaml Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
The uv.lock is gitignored so Shipwright builds fail with --locked. Remove flag to let uv resolve deps at build time. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Add one-command deployment script and Kubernetes manifest for deploying kagenti_chat to a Kagenti cluster, integrating with llm-d for inference. The deployment uses: - Shipwright buildah-insecure-direct ClusterBuildStrategy for in-cluster builds - Internal registry via ClusterIP (10.43.28.116:5000) to bypass DNS issues - Kagenti shared HTTP gateway (kagenti-system/http) for external access - Direct llm-d service URL for connecting to Llama-3.3-70B-Instruct - kagenti.io/inject=disabled label to skip authbridge sidecar injection Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Document the deploy-to-kagenti.sh script, prerequisites, and verification steps. Include architecture notes explaining the cluster-specific workarounds (ClusterIP image reference, kagenti-system gateway, direct llm-d service URL, authbridge injection skip). Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Add a scrolling terminal CLI for interacting with the deployed kagenti_chat agent over A2A JSON-RPC. Lives at src/kagenti_chat/client/cli.py and is exposed via the `kagenti-chat` console script. Features: - Streaming responses with fallback to non-streaming - Markdown rendering for code blocks, lists, and headers - Conversation history persisted to ~/.kagenti-chat/history.jsonl - Input history with up/down navigation - Slash commands: /help, /clear, /save, /info, /exit The client defaults to the deployed cluster URL so colleagues can install and run with no configuration. Override via --url or KAGENTI_CHAT_URL. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
The deployed cluster uses a self-signed cert, which made first-run UX require knowing about --insecure. Detect cert errors during initial connect and retry without verification, with a one-line notice. Also fix a "Cannot open a client instance more than once" error from the connect refactor by using try/finally to close the httpx.AsyncClient instead of async-with. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Add optional API_KEY-based bearer token authentication to the agent server and CLI client. Server (kagenti_chat.a2a_server): - New BearerAuthMiddleware enforces Authorization: Bearer header on all paths except /.well-known/agent.json and /.well-known/agent-card.json, which must remain public for client discovery. - Uses secrets.compare_digest to avoid timing attacks. - Disabled when API_KEY is unset; logs a warning so it's obvious. Client (kagenti-chat CLI): - Accepts --api-key flag and KAGENTI_CHAT_API_KEY env var. - Sends Authorization: Bearer on message/send and message/stream calls. - Maps 401/403 responses to friendly error messages. Deployment: - deploy-to-kagenti.sh generates a random key and stores it in a kagenti-chat-api-key Secret on first run. - k8s manifest reads API_KEY from the Secret with optional: true so the deployment still works without authentication if the Secret is absent. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Add an Authentication section to the README covering how to retrieve the deployed key, rotate it, and what happens when API_KEY is unset. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
A minimal A2A agent for validating durable agent context on Kagenti. One-node LangGraph chat agent with two persistence paths: - Per-turn JSONL appended to <CONTEXT_DIR>/<sanitized-context-id>.jsonl. Default CONTEXT_DIR is /shared/minimal-pvc-agent, the Kagenti StatefulSet PVC mount path. Best-effort: IO errors are logged and swallowed so the agent stays responsive when /shared is unavailable. - LangGraph checkpointer keyed on the A2A context_id as thread_id. MemorySaver by default; AsyncPostgresSaver when CHECKPOINT_DB_URL is set, so conversation state can survive pod restarts. A2A task storage mirrors the same shape: InMemoryTaskStore by default, DatabaseTaskStore (a2a-sdk[sql], SQLAlchemy async engine) when TASK_STORE_DB_URL is set. Two read-only HTTP endpoints inspect persistence without `kubectl exec`: - GET /history?context_id=... returns the JSONL turn log - GET /checkpoint?context_id=... returns the LangGraph MessagesState Falls back to echo mode (returns "echo: <input> (turn N)") when no valid LLM credentials are configured, so smoke tests work without an LLM. Uses the same has_valid_api_key heuristic as weather_service: dummy keys are accepted only when the LLM base URL is localhost. a2a-sdk pinned to >=0.3.26,<1.0 to match the version running on the target cluster (verified via kubectl exec against the live weather agent). Includes 24 unit tests covering configuration defaults and env overrides, has_valid_api_key heuristic, context-id sanitization (path traversal, length cap, unicode), and JSONL append/read (append, separate contexts, missing root mkdir, swallowed OSError, malformed-line skipping). End-to-end verified locally: two-turn A2A conversation produced two JSONL records, /history returned both turns, /checkpoint returned a four-message LangGraph state proving cross-turn memory. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
Replace the custom JSONL writer with the harness's own persistence primitives, both backed by SQLite files on the PVC: - LangGraph AsyncSqliteSaver at CHECKPOINT_PATH (default /shared/checkpoints.db). Empty string falls back to MemorySaver. - A2A DatabaseTaskStore via SQLAlchemy asyncio over a SQLite file at TASK_STORE_PATH (default /shared/tasks.db). Empty string falls back to InMemoryTaskStore. The agent code no longer contains any persistence logic. Removing the JSONL writer (persistence.py, /history endpoint, append_turn call in execute()) also removes the burden of every agent author having to reimplement basic session persistence: the harness already knows how to do it, we just need to point it at durable storage. Drops langgraph-checkpoint-postgres and asyncpg dependencies; adds langgraph-checkpoint-sqlite and aiosqlite. Postgres support can be reintroduced later when we need multi-writer semantics. Smoke-tested locally: two turns in one context produced a checkpoints.db containing four messages (2 human + 2 AI) via GET /checkpoint. The turn counter correctly incremented to (turn 2), proving LangGraph saw turn 1's messages via the checkpointer. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
The Kagenti CLI's `kagenti chat` calls `message/stream` by default, but the agent card previously declared `streaming=false`, so the a2a-sdk rejected the request with "Streaming is not supported by the agent". Enable streaming by: - Declaring `AgentCapabilities(streaming=True)` in the agent card. - Replacing `graph.ainvoke(...)` with `graph.astream(..., stream_mode="updates")` so intermediate LangGraph node updates are visible to clients. - Emitting an initial `TaskState.working` "thinking..." status before the graph runs so streaming clients see progress right away. - Emitting one working-status message per node event, with node output summaries truncated to 256 chars (same convention as weather_service). The final artifact + `input_required` status logic is unchanged, so non-streaming `message/send` clients still get the same response shape as before. The persistence layer (SQLite-on-PVC checkpointer, task store) is untouched. Bumps the agent version from 0.2.0 to 0.3.0. Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com> Signed-off-by: Jeremy Cohn <Jeremy.Cohn@ibm.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
kagenti chat minimal-pvc-agentwas failing with"Streaming is not supported by the agent"because the agent card declaredstreaming=falsewhile the CLI's default path callsmessage/stream. This PR adds real streaming support so the CLI's default path just works.Changes
AgentCapabilities(streaming=True)in the agent card.graph.ainvoke(...)withgraph.astream(..., stream_mode="updates")inMinimalPVCExecutor.executeso intermediate LangGraph node events are surfaced.TaskState.working"thinking..." status before the graph runs, then one working-status per node event with output summaries truncated to 256 chars (same convention asweather_service).input_requiredstatus logic unchanged, so non-streamingmessage/sendclients still get the same response shape.Testing
uv run --with pytest --with pydantic-settings --python 3.11 python -m pytest tests/a2a/test_minimal_pvc_agent.py -q— 9 passed.curl /.well-known/agent-card.jsonshows"streaming": true,"version": "0.3.0".message/sendreturns the echo reply with the artifact (backward compatible).message/streamreturns SSE with initial status → intermediate node events → artifact → terminal status.Related
Complements upstream Kagenti PR that teaches
kagenti chatto fall back tomessage/sendfor agents that declarestreaming=false. Either fix alone unblockskagenti chatfor this agent; both together are the belt-and-suspenders story.