feat(minimal_pvc_agent): support A2A streaming - #1
Open
moonlight16 wants to merge 3 commits into
Open
Conversation
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.