From f853785aaf4d02f3ce694151193fbbe4020a233a Mon Sep 17 00:00:00 2001 From: Michael Xu Date: Thu, 10 Sep 2026 12:15:01 -0500 Subject: [PATCH] feat(harness): add a Gemini CLI harness (tap, turn, and init templates) Adds Gemini CLI as a framework harness alongside Claude Code and Codex: - convert_gemini_cli_to_agentex_events: maps the CLI's stream-json events (init, message deltas, tool_use, tool_result, error, result; schema per packages/core/src/output/types.ts in google-gemini/gemini-cli) onto the canonical StreamTaskMessage* stream. Assistant deltas open one text slot that closes on the next tool event, the result, or end of stream, so every Start has a Done; tool requests and results pair by tool_id. - GeminiCliTurn: HarnessTurn wrapper exposing session_id and model from the init event and normalising result.stats into TurnUsage. - Both exported from agentex.lib.adk. - agentex init templates sync-gemini-cli, default-gemini-cli and temporal-gemini-cli (registered in TemplateType, file map and menus), cloned from the Claude Code templates: prompt passed via -p with stdin closed (the CLI reads stdin to EOF in headless mode), optional GEMINI_MODEL, GEMINI_API_KEY credential, npm install -g @google/gemini-cli in the Dockerfile. Turns are independent prompts: the CLI's --resume takes latest/index, not a session id. - Tests: tap (text deltas, whole messages, tools, errors, callbacks, source close on cancel), turn (usage mapping, protocol), harness end to end through UnifiedEmitter with span derivation; template suite covers the three new templates. Offline tests only; a live smoke run needs a Gemini API key. Claude-Session: https://claude.ai/code/session_01HCVKnA7LeJZ44nxZz1uzF3 --- src/agentex/lib/adk/__init__.py | 6 + .../lib/adk/_modules/_gemini_cli_sync.py | 276 ++++++++++++++++++ .../lib/adk/_modules/_gemini_cli_turn.py | 139 +++++++++ src/agentex/lib/cli/commands/init.py | 9 + .../default-gemini-cli/.dockerignore.j2 | 43 +++ .../default-gemini-cli/.env.example.j2 | 13 + .../default-gemini-cli/Dockerfile-uv.j2 | 51 ++++ .../default-gemini-cli/Dockerfile.j2 | 46 +++ .../templates/default-gemini-cli/README.md.j2 | 64 ++++ .../templates/default-gemini-cli/dev.ipynb.j2 | 126 ++++++++ .../default-gemini-cli/environments.yaml.j2 | 57 ++++ .../default-gemini-cli/manifest.yaml.j2 | 123 ++++++++ .../default-gemini-cli/project/acp.py.j2 | 167 +++++++++++ .../default-gemini-cli/pyproject.toml.j2 | 33 +++ .../default-gemini-cli/requirements.txt.j2 | 8 + .../sync-gemini-cli/.dockerignore.j2 | 43 +++ .../templates/sync-gemini-cli/.env.example.j2 | 13 + .../sync-gemini-cli/Dockerfile-uv.j2 | 51 ++++ .../templates/sync-gemini-cli/Dockerfile.j2 | 47 +++ .../templates/sync-gemini-cli/README.md.j2 | 64 ++++ .../templates/sync-gemini-cli/dev.ipynb.j2 | 167 +++++++++++ .../sync-gemini-cli/environments.yaml.j2 | 53 ++++ .../sync-gemini-cli/manifest.yaml.j2 | 120 ++++++++ .../sync-gemini-cli/project/acp.py.j2 | 155 ++++++++++ .../sync-gemini-cli/pyproject.toml.j2 | 33 +++ .../sync-gemini-cli/requirements.txt.j2 | 8 + .../temporal-gemini-cli/.dockerignore.j2 | 43 +++ .../temporal-gemini-cli/.env.example.j2 | 13 + .../temporal-gemini-cli/Dockerfile-uv.j2 | 61 ++++ .../temporal-gemini-cli/Dockerfile.j2 | 54 ++++ .../temporal-gemini-cli/README.md.j2 | 72 +++++ .../temporal-gemini-cli/dev.ipynb.j2 | 126 ++++++++ .../temporal-gemini-cli/environments.yaml.j2 | 64 ++++ .../temporal-gemini-cli/manifest.yaml.j2 | 142 +++++++++ .../temporal-gemini-cli/project/acp.py.j2 | 31 ++ .../project/activities.py.j2 | 156 ++++++++++ .../project/run_worker.py.j2 | 41 +++ .../project/workflow.py.j2 | 149 ++++++++++ .../temporal-gemini-cli/pyproject.toml.j2 | 37 +++ .../temporal-gemini-cli/requirements.txt.j2 | 11 + tests/lib/adk/test_gemini_cli_sync.py | 192 ++++++++++++ tests/lib/adk/test_gemini_cli_turn.py | 121 ++++++++ .../harness/test_harness_gemini_cli_sync.py | 98 +++++++ 43 files changed, 3326 insertions(+) create mode 100644 src/agentex/lib/adk/_modules/_gemini_cli_sync.py create mode 100644 src/agentex/lib/adk/_modules/_gemini_cli_turn.py create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/.dockerignore.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/.env.example.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile-uv.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/README.md.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/dev.ipynb.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/environments.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/manifest.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/project/acp.py.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/pyproject.toml.j2 create mode 100644 src/agentex/lib/cli/templates/default-gemini-cli/requirements.txt.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/.dockerignore.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/.env.example.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile-uv.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/README.md.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/dev.ipynb.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/environments.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/manifest.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/project/acp.py.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/pyproject.toml.j2 create mode 100644 src/agentex/lib/cli/templates/sync-gemini-cli/requirements.txt.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/.dockerignore.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/.env.example.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile-uv.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/README.md.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/dev.ipynb.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/environments.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/manifest.yaml.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/project/acp.py.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/project/activities.py.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/project/run_worker.py.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/project/workflow.py.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/pyproject.toml.j2 create mode 100644 src/agentex/lib/cli/templates/temporal-gemini-cli/requirements.txt.j2 create mode 100644 tests/lib/adk/test_gemini_cli_sync.py create mode 100644 tests/lib/adk/test_gemini_cli_turn.py create mode 100644 tests/lib/core/harness/test_harness_gemini_cli_sync.py diff --git a/src/agentex/lib/adk/__init__.py b/src/agentex/lib/adk/__init__.py index c05f8f3ea..bfa2422ed 100644 --- a/src/agentex/lib/adk/__init__.py +++ b/src/agentex/lib/adk/__init__.py @@ -22,6 +22,8 @@ ) from agentex.lib.adk._modules._codex_sync import convert_codex_to_agentex_events from agentex.lib.adk._modules._codex_turn import CodexTurn, codex_usage_to_turn_usage +from agentex.lib.adk._modules._gemini_cli_sync import convert_gemini_cli_to_agentex_events +from agentex.lib.adk._modules._gemini_cli_turn import GeminiCliTurn, gemini_cli_usage_to_turn_usage from agentex.lib.adk._modules.events import EventsModule from agentex.lib.adk._modules.messages import MessagesModule from agentex.lib.adk._modules.state import StateModule @@ -101,6 +103,10 @@ "convert_codex_to_agentex_events", "CodexTurn", "codex_usage_to_turn_usage", + # Gemini CLI + "convert_gemini_cli_to_agentex_events", + "GeminiCliTurn", + "gemini_cli_usage_to_turn_usage", # Unified harness surface (AGX1-375) "UnifiedEmitter", "SpanTracer", diff --git a/src/agentex/lib/adk/_modules/_gemini_cli_sync.py b/src/agentex/lib/adk/_modules/_gemini_cli_sync.py new file mode 100644 index 000000000..0e523f0b5 --- /dev/null +++ b/src/agentex/lib/adk/_modules/_gemini_cli_sync.py @@ -0,0 +1,276 @@ +"""Gemini CLI stream-json parser tap for the unified harness surface. + +Converts the newline-delimited JSON events emitted by +``gemini -p --output-format stream-json`` into the canonical +``StreamTaskMessage*`` stream consumed by the Agentex harness. + +Event → canonical mapping +------------------------- +init + Fires ``on_init`` with the raw event (``session_id``, ``model``). Nothing + is emitted: session metadata is a provider concern. + +message (role=user) + Ignored. The CLI echoes the prompt back as the first message. + +message (role=assistant) + The CLI streams the answer as ``delta: true`` chunks. The first chunk + opens a text slot (Start(TextContent)); every chunk is a Delta(TextDelta). + The slot is closed (Done) when a ``tool_use``, ``tool_result`` or + ``result`` event arrives, or when the stream ends. A non-delta assistant + message whose content matches the open slot closes it; otherwise it is + delivered as Start + Delta + Done. + +tool_use + Start(ToolRequestContent) + Done. ``tool_id`` → ``tool_call_id``, + ``tool_name`` → ``name``, ``parameters`` → ``arguments``. + +tool_result + Full(ToolResponseContent) keyed by ``tool_id``. ``output`` (or the error + message when ``status == "error"``) becomes ``content["result"]``; + ``is_error`` is set for error results. + +error + Logged (``severity`` + ``message``). Nothing is emitted. + +result + Closes any open text slot, then fires ``on_result`` with the raw event so + the caller can read ``stats`` (tokens, duration, tool calls). + +Reference: ``packages/core/src/output/types.ts`` in google-gemini/gemini-cli. +""" + +from __future__ import annotations + +import json +from typing import Any, Callable, Awaitable, AsyncIterator + +from agentex.lib.utils.logging import make_logger +from agentex.types.text_content import TextContent +from agentex.types.task_message_delta import TextDelta +from agentex.types.task_message_update import ( + StreamTaskMessageDone, + StreamTaskMessageFull, + StreamTaskMessageDelta, + StreamTaskMessageStart, +) +from agentex.types.tool_request_content import ToolRequestContent +from agentex.types.tool_response_content import ToolResponseContent + +logger = make_logger(__name__) + +_MAX_RESULT_LENGTH = 4000 + + +def _truncate(text: str) -> str: + return str(text)[:_MAX_RESULT_LENGTH] + + +async def convert_gemini_cli_to_agentex_events( + lines: AsyncIterator[str | dict[str, Any]], + on_result: Callable[[dict[str, Any]], Awaitable[None]] | None = None, + on_init: Callable[[dict[str, Any]], Awaitable[None]] | None = None, +) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]: + """Public tap: convert a Gemini CLI ``stream-json`` line stream to events. + + Thin wrapper over :func:`_convert_gemini_cli_impl` that owns the + cancellation backstop: a ``finally`` closes the underlying ``lines`` + iterator (when it exposes ``aclose``) whenever this generator is closed, + including on the ``GeneratorExit``/``CancelledError`` raised when the + consuming task is cancelled mid-turn, so the CLI stdout handle and + subprocess are not leaked. + """ + inner = _convert_gemini_cli_impl(lines, on_result=on_result, on_init=on_init) + try: + async for event in inner: + yield event + finally: + inner_aclose = getattr(inner, "aclose", None) + if inner_aclose is not None: + await inner_aclose() + aclose = getattr(lines, "aclose", None) + if aclose is not None: + await aclose() + + +async def _convert_gemini_cli_impl( + lines: AsyncIterator[str | dict[str, Any]], + on_result: Callable[[dict[str, Any]], Awaitable[None]] | None = None, + on_init: Callable[[dict[str, Any]], Awaitable[None]] | None = None, +) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]: + """Convert a Gemini CLI ``stream-json`` line stream into ``StreamTaskMessage*`` events. + + Each item in ``lines`` is either a raw JSON string (as read from the CLI's + stdout) or an already-parsed dict. Empty strings are skipped; unparseable + JSON is logged and skipped. The event → canonical mapping is documented in + this module's docstring. + """ + next_index = 0 + tool_call_count = 0 + + # One open assistant text slot at a time: the CLI streams the answer as + # ``delta: true`` message chunks with no explicit start/stop markers. + text_open = False + text_index: int | None = None + text_buf = "" + + def _close_text() -> StreamTaskMessageDone | None: + nonlocal text_open, text_index, text_buf + if not text_open or text_index is None: + return None + done = StreamTaskMessageDone(type="done", index=text_index) + text_open = False + text_index = None + text_buf = "" + return done + + async for raw in lines: + if not raw: + continue + + if isinstance(raw, dict): + evt = raw + else: + line = raw.strip() + if not line: + continue + try: + evt = json.loads(line) + except json.JSONDecodeError: + logger.debug("gemini-cli: skipping non-JSON line: %r", line[:120]) + continue + + if not isinstance(evt, dict): + continue + evt_type = evt.get("type", "") + + if evt_type == "message": + if evt.get("role") != "assistant": + continue # the CLI echoes the user prompt; nothing to emit + content = evt.get("content", "") + if not isinstance(content, str) or not content: + continue + + if evt.get("delta"): + if not text_open: + text_open = True + text_index = next_index + next_index += 1 + text_buf = "" + yield StreamTaskMessageStart( + type="start", + index=text_index, + content=TextContent(type="text", author="agent", content=""), + ) + text_buf += content + assert text_index is not None + yield StreamTaskMessageDelta( + type="delta", + index=text_index, + delta=TextDelta(type="text", text_delta=content), + ) + continue + + # A complete (non-delta) assistant message. If it materialises the + # slot we are already streaming, just close the slot; otherwise + # deliver it as its own Start + Delta + Done. + if text_open and text_buf and content.startswith(text_buf): + done = _close_text() + if done is not None: + yield done + continue + done = _close_text() + if done is not None: + yield done + msg_index = next_index + next_index += 1 + yield StreamTaskMessageStart( + type="start", + index=msg_index, + content=TextContent(type="text", author="agent", content=""), + ) + yield StreamTaskMessageDelta( + type="delta", + index=msg_index, + delta=TextDelta(type="text", text_delta=content), + ) + yield StreamTaskMessageDone(type="done", index=msg_index) + + elif evt_type == "tool_use": + done = _close_text() + if done is not None: + yield done + tool_call_count += 1 + tool_id = evt.get("tool_id") or f"tool_{tool_call_count}" + name = evt.get("tool_name") or "unknown" + arguments = evt.get("parameters") + if not isinstance(arguments, dict): + arguments = {} + msg_index = next_index + next_index += 1 + yield StreamTaskMessageStart( + type="start", + index=msg_index, + content=ToolRequestContent( + type="tool_request", + author="agent", + tool_call_id=str(tool_id), + name=str(name), + arguments=arguments, + ), + ) + yield StreamTaskMessageDone(type="done", index=msg_index) + + elif evt_type == "tool_result": + done = _close_text() + if done is not None: + yield done + tool_id = str(evt.get("tool_id") or "") + is_error = evt.get("status") == "error" + output = evt.get("output") + if output is None: + error = evt.get("error") or {} + output = error.get("message", "") if isinstance(error, dict) else str(error) + result_content: dict[str, Any] = {"result": _truncate(str(output))} + if is_error: + result_content["is_error"] = True + msg_index = next_index + next_index += 1 + yield StreamTaskMessageFull( + type="full", + index=msg_index, + content=ToolResponseContent( + type="tool_response", + author="agent", + tool_call_id=tool_id, + name="", + content=result_content, + ), + ) + + elif evt_type == "init": + if on_init is not None: + await on_init(evt) + + elif evt_type == "error": + logger.warning( + "gemini-cli: %s: %s", + evt.get("severity", "error"), + str(evt.get("message", ""))[:300], + ) + + elif evt_type == "result": + done = _close_text() + if done is not None: + yield done + if on_result is not None: + await on_result(evt) + + else: + logger.debug("gemini-cli: unhandled event type %r", evt_type) + + # Stream ended without a result event (truncated / interrupted): close the + # slot so every Start has a matching Done. + done = _close_text() + if done is not None: + yield done diff --git a/src/agentex/lib/adk/_modules/_gemini_cli_turn.py b/src/agentex/lib/adk/_modules/_gemini_cli_turn.py new file mode 100644 index 000000000..1abfc4a64 --- /dev/null +++ b/src/agentex/lib/adk/_modules/_gemini_cli_turn.py @@ -0,0 +1,139 @@ +"""GeminiCliTurn — HarnessTurn implementation for the Gemini CLI tap. + +Wraps ``convert_gemini_cli_to_agentex_events`` to implement the +``HarnessTurn`` protocol: exposes ``events`` (the canonical +``StreamTaskMessage*`` stream) and ``usage()`` (the normalised ``TurnUsage``, +populated after the stream is exhausted). + +Usage normalization +------------------- +The CLI's terminal ``result`` event carries ``stats``: + + stats.input_tokens -> input_tokens + stats.output_tokens -> output_tokens + stats.cached -> cached_input_tokens + stats.total_tokens -> total_tokens (or input + output when absent) + stats.duration_ms -> duration_ms + stats.tool_calls -> num_tool_calls + init.model / stats.models -> model + +The CLI does not report cost or the number of model calls, so ``cost_usd`` +and ``num_llm_calls`` stay ``None``. Real zeros are preserved; missing keys +default to ``None`` so consumers can tell "not reported" from "zero". +""" + +from __future__ import annotations + +from typing import Any, AsyncIterator + +from agentex.lib.core.harness.types import TurnUsage, HarnessTurn, StreamTaskMessage +from agentex.lib.adk._modules._gemini_cli_sync import convert_gemini_cli_to_agentex_events + + +def gemini_cli_usage_to_turn_usage(result_envelope: dict[str, Any], model: str | None = None) -> TurnUsage: + """Map a Gemini CLI ``result`` event to a canonical ``TurnUsage``. + + ``model`` (from the ``init`` event) wins; otherwise the first model named + under ``stats.models`` is used. Missing values map to ``None``. + """ + stats: dict[str, Any] = result_envelope.get("stats") or {} + + def _int(d: dict[str, Any], key: str) -> int | None: + v = d.get(key) + if v is None: + return None + try: + return int(v) + except (TypeError, ValueError): + return None + + input_tokens = _int(stats, "input_tokens") + output_tokens = _int(stats, "output_tokens") + cached_input_tokens = _int(stats, "cached") + total_tokens = _int(stats, "total_tokens") + if total_tokens is None and input_tokens is not None and output_tokens is not None: + total_tokens = input_tokens + output_tokens + duration_ms = _int(stats, "duration_ms") + num_tool_calls = _int(stats, "tool_calls") or 0 + + if model is None: + models = stats.get("models") + if isinstance(models, dict) and models: + model = next(iter(models)) + + return TurnUsage( + model=model, + input_tokens=input_tokens, + output_tokens=output_tokens, + cached_input_tokens=cached_input_tokens, + total_tokens=total_tokens, + duration_ms=duration_ms, + num_tool_calls=num_tool_calls, + ) + + +class GeminiCliTurn: + """HarnessTurn for a Gemini CLI ``stream-json`` line stream. + + Satisfies the ``HarnessTurn`` protocol: + - ``events`` yields the canonical ``StreamTaskMessage*`` stream. + - ``usage()`` returns the normalised ``TurnUsage`` (only valid after + ``events`` is fully consumed). + + ``lines`` is an async iterator of raw JSON strings or pre-parsed dicts, as + produced by reading the ``gemini`` CLI's stdout line by line. + """ + + def __init__(self, lines: AsyncIterator[str | dict[str, Any]]) -> None: + self._lines = lines + self._result_envelope: dict[str, Any] | None = None + self._session_id: str | None = None + self._model: str | None = None + self._events_stream: AsyncIterator[StreamTaskMessage] | None = None + + async def _on_result(self, envelope: dict[str, Any]) -> None: + self._result_envelope = envelope + + async def _on_init(self, envelope: dict[str, Any]) -> None: + sid = envelope.get("session_id") + if sid: + self._session_id = str(sid) + model = envelope.get("model") + if model: + self._model = str(model) + + @property + def events(self) -> AsyncIterator[StreamTaskMessage]: + if self._events_stream is None: + self._events_stream = convert_gemini_cli_to_agentex_events( + self._lines, + on_result=self._on_result, + on_init=self._on_init, + ) + return self._events_stream + + @property + def session_id(self) -> str | None: + """The Gemini CLI session id from the ``init`` event, if reported.""" + return self._session_id + + @property + def model(self) -> str | None: + """The model name from the ``init`` event, if reported.""" + return self._model + + def usage(self) -> TurnUsage: + """Return normalised usage for this turn. + + Call only after ``events`` is exhausted. Returns an empty ``TurnUsage`` + if the ``result`` event was not received (e.g. the stream was truncated). + """ + if self._result_envelope is None: + return TurnUsage(model=self._model) + return gemini_cli_usage_to_turn_usage(self._result_envelope, model=self._model) + + +# Runtime assert that GeminiCliTurn satisfies the HarnessTurn protocol +assert isinstance(GeminiCliTurn.__new__(GeminiCliTurn), HarnessTurn), ( + "GeminiCliTurn must satisfy the HarnessTurn protocol" +) diff --git a/src/agentex/lib/cli/commands/init.py b/src/agentex/lib/cli/commands/init.py index 9849e9bbc..2b4e26380 100644 --- a/src/agentex/lib/cli/commands/init.py +++ b/src/agentex/lib/cli/commands/init.py @@ -28,12 +28,14 @@ class TemplateType(str, Enum): TEMPORAL_LANGGRAPH = "temporal-langgraph" TEMPORAL_CLAUDE_CODE = "temporal-claude-code" TEMPORAL_CODEX = "temporal-codex" + TEMPORAL_GEMINI_CLI = "temporal-gemini-cli" DEFAULT = "default" DEFAULT_LANGGRAPH = "default-langgraph" DEFAULT_PYDANTIC_AI = "default-pydantic-ai" DEFAULT_OPENAI_AGENTS = "default-openai-agents" DEFAULT_CLAUDE_CODE = "default-claude-code" DEFAULT_CODEX = "default-codex" + DEFAULT_GEMINI_CLI = "default-gemini-cli" SYNC = "sync" SYNC_OPENAI_AGENTS = "sync-openai-agents" SYNC_OPENAI_AGENTS_LOCAL_SANDBOX = "sync-openai-agents-local-sandbox" @@ -41,6 +43,7 @@ class TemplateType(str, Enum): SYNC_PYDANTIC_AI = "sync-pydantic-ai" SYNC_CLAUDE_CODE = "sync-claude-code" SYNC_CODEX = "sync-codex" + SYNC_GEMINI_CLI = "sync-gemini-cli" def render_template( @@ -75,12 +78,14 @@ def create_project_structure( TemplateType.TEMPORAL_LANGGRAPH: ["acp.py", "workflow.py", "run_worker.py", "graph.py", "tools.py"], TemplateType.TEMPORAL_CLAUDE_CODE: ["acp.py", "workflow.py", "run_worker.py", "activities.py"], TemplateType.TEMPORAL_CODEX: ["acp.py", "workflow.py", "run_worker.py", "activities.py"], + TemplateType.TEMPORAL_GEMINI_CLI: ["acp.py", "workflow.py", "run_worker.py", "activities.py"], TemplateType.DEFAULT: ["acp.py"], TemplateType.DEFAULT_LANGGRAPH: ["acp.py", "graph.py", "tools.py"], TemplateType.DEFAULT_PYDANTIC_AI: ["acp.py", "agent.py", "tools.py"], TemplateType.DEFAULT_OPENAI_AGENTS: ["acp.py"], TemplateType.DEFAULT_CLAUDE_CODE: ["acp.py"], TemplateType.DEFAULT_CODEX: ["acp.py"], + TemplateType.DEFAULT_GEMINI_CLI: ["acp.py"], TemplateType.SYNC: ["acp.py"], TemplateType.SYNC_OPENAI_AGENTS: ["acp.py"], TemplateType.SYNC_OPENAI_AGENTS_LOCAL_SANDBOX: ["acp.py", "agent.py", "tools.py"], @@ -88,6 +93,7 @@ def create_project_structure( TemplateType.SYNC_PYDANTIC_AI: ["acp.py", "agent.py", "tools.py"], TemplateType.SYNC_CLAUDE_CODE: ["acp.py"], TemplateType.SYNC_CODEX: ["acp.py"], + TemplateType.SYNC_GEMINI_CLI: ["acp.py"], }[template_type] # Create project/code files @@ -203,6 +209,7 @@ def validate_agent_name(text: str) -> bool | str: {"name": "Async ACP + Pydantic AI", "value": TemplateType.DEFAULT_PYDANTIC_AI}, {"name": "Async ACP + Claude Code", "value": TemplateType.DEFAULT_CLAUDE_CODE}, {"name": "Async ACP + Codex", "value": TemplateType.DEFAULT_CODEX}, + {"name": "Async ACP + Gemini CLI", "value": TemplateType.DEFAULT_GEMINI_CLI}, ], ).ask() if not template_type: @@ -217,6 +224,7 @@ def validate_agent_name(text: str) -> bool | str: {"name": "Temporal + LangGraph", "value": TemplateType.TEMPORAL_LANGGRAPH}, {"name": "Temporal + Claude Code", "value": TemplateType.TEMPORAL_CLAUDE_CODE}, {"name": "Temporal + Codex", "value": TemplateType.TEMPORAL_CODEX}, + {"name": "Temporal + Gemini CLI", "value": TemplateType.TEMPORAL_GEMINI_CLI}, ], ).ask() if not template_type: @@ -232,6 +240,7 @@ def validate_agent_name(text: str) -> bool | str: {"name": "Sync ACP + Pydantic AI", "value": TemplateType.SYNC_PYDANTIC_AI}, {"name": "Sync ACP + Claude Code", "value": TemplateType.SYNC_CLAUDE_CODE}, {"name": "Sync ACP + Codex", "value": TemplateType.SYNC_CODEX}, + {"name": "Sync ACP + Gemini CLI", "value": TemplateType.SYNC_GEMINI_CLI}, ], ).ask() if not template_type: diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/.dockerignore.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/.dockerignore.j2 new file mode 100644 index 000000000..c2d7fca4d --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/.dockerignore.j2 @@ -0,0 +1,43 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Environments +.env** +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Git +.git +.gitignore + +# Misc +.DS_Store diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/.env.example.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/.env.example.j2 new file mode 100644 index 000000000..611d7886b --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/.env.example.j2 @@ -0,0 +1,13 @@ +# {{ agent_name }} - Environment Variables +# Copy this file to .env and fill in the values + +# API key for the Gemini CLI (the `gemini` subprocess this agent spawns) +GEMINI_API_KEY= + +# LLM base URL (optional - override to use a different provider) +# OPENAI_BASE_URL= + +# SGP Configuration (optional - for tracing) +# SGP_API_KEY= +# SGP_ACCOUNT_ID= +# SGP_CLIENT_BASE_URL= diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile-uv.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile-uv.j2 new file mode 100644 index 000000000..0f9880ac0 --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile-uv.j2 @@ -0,0 +1,51 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/** + +# Install the Gemini CLI CLI: the agent shells out to `gemini` on every turn, +# so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +ENV UV_COMPILE_BYTECODE=1 +ENV UV_LINK_MODE=copy +ENV UV_HTTP_TIMEOUT=1000 + +WORKDIR /app/{{ project_path_from_build_root }} + +# Copy dependency files for layer caching +COPY {{ project_path_from_build_root }}/pyproject.toml ./ + +# Install dependencies (without project itself, for layer caching) +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-install-project --no-dev + +# Copy the project code +COPY {{ project_path_from_build_root }}/project ./project + +# Install the project +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-dev + +ENV PATH="/app/{{ project_path_from_build_root }}/.venv/bin:$PATH" +ENV PYTHONPATH=/app + +# Run the agent using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile.j2 new file mode 100644 index 000000000..668014b47 --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/Dockerfile.j2 @@ -0,0 +1,46 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/* + +# Install the Gemini CLI CLI: the agent shells out to `gemini` on every turn, +# so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +RUN uv pip install --system --upgrade pip setuptools wheel + +ENV UV_HTTP_TIMEOUT=1000 + +# Copy just the requirements file to optimize caching +COPY {{ project_path_from_build_root }}/requirements.txt /app/{{ project_path_from_build_root }}/requirements.txt + +WORKDIR /app/{{ project_path_from_build_root }} + +# Install the required Python packages +RUN uv pip install --system -r requirements.txt + +# Copy the project code +COPY {{ project_path_from_build_root }}/project /app/{{ project_path_from_build_root }}/project + +# Set environment variables +ENV PYTHONPATH=/app + +# Run the agent using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/README.md.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/README.md.j2 new file mode 100644 index 000000000..ffcfee62b --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/README.md.j2 @@ -0,0 +1,64 @@ +# {{ agent_name }} - AgentEx Async Gemini CLI Agent + +This template builds an **asynchronous** (non-Temporal) agent that drives the +**Gemini CLI CLI** through the unified harness surface on AgentEx: +- Spawns `gemini -p "" --output-format stream-json` as a local subprocess +- Wraps the CLI's stdout stream in a `GeminiCliTurn` +- Delivers canonical `StreamTaskMessage*` events via `UnifiedEmitter.auto_send_turn` + (the async Redis push path), so the UI receives output in real time +- Tracing integration to SGP / AgentEx + +## Prerequisites + +- The `gemini` CLI installed and on your `PATH` +- An `GEMINI_API_KEY` (or equivalent credential) in your environment + +## Running the Agent + +```bash +agentex agents run --manifest manifest.yaml +``` + +## Project Structure + +``` +{{ project_name }}/ +├── project/ +│ ├── __init__.py +│ └── acp.py # ACP server, subprocess spawn, and event handlers +├── Dockerfile +├── manifest.yaml +├── dev.ipynb +{% if use_uv %} +└── pyproject.toml +{% else %} +└── requirements.txt +{% endif %} +``` + +## Key Concepts + +### Async ACP with the harness +The async ACP model streams events over Redis instead of an HTTP response. The +`@acp.on_task_event_send` handler spawns the Gemini CLI CLI and pushes the +harness events to the task stream. + +### The unified harness surface +`GeminiCliTurn` + `UnifiedEmitter` are the unified harness surface. The turn +normalizes CLI output into canonical AgentEx events; the emitter traces and +delivers them. + +## Development + +### 1. Customize the subprocess +Edit `_spawn_gemini` in `project/acp.py` to change the CLI flags, working +directory, or how the prompt is delivered. + +### 2. Configure Credentials +Set your credentials via `manifest.yaml`, an exported environment variable, or a +`.env` file in the project directory. + +### 3. Run Locally +```bash +export ENVIRONMENT=development && agentex agents run --manifest manifest.yaml +``` diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/dev.ipynb.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/dev.ipynb.j2 new file mode 100644 index 000000000..d3a68303f --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/dev.ipynb.j2 @@ -0,0 +1,126 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "36834357", + "metadata": {}, + "outputs": [], + "source": [ + "from agentex import Agentex\n", + "\n", + "client = Agentex(base_url=\"http://localhost:5003\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d1c309d6", + "metadata": {}, + "outputs": [], + "source": [ + "AGENT_NAME = \"{{ agent_name }}\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9f6e6ef0", + "metadata": {}, + "outputs": [], + "source": [ + "# (REQUIRED) Create a new task. For Async agents, you must create a task for messages to be associated with.\n", + "import uuid\n", + "\n", + "rpc_response = client.agents.create_task(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"name\": f\"{str(uuid.uuid4())[:8]}-task\",\n", + " \"params\": {}\n", + " }\n", + ")\n", + "\n", + "task = rpc_response.result\n", + "print(task)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b03b0d37", + "metadata": {}, + "outputs": [], + "source": [ + "# Send an event to the agent\n", + "\n", + "# The response is expected to be a list of TaskMessage objects, which is a union of the following types:\n", + "# - TextContent: A message with just text content \n", + "# - DataContent: A message with JSON-serializable data content\n", + "# - ToolRequestContent: A message with a tool request, which contains a JSON-serializable request to call a tool\n", + "# - ToolResponseContent: A message with a tool response, which contains response object from a tool call in its content\n", + "\n", + "# When processing the message/send response, if you are expecting more than TextContent, such as DataContent, ToolRequestContent, or ToolResponseContent, you can process them as well\n", + "\n", + "rpc_response = client.agents.send_event(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"content\": {\"type\": \"text\", \"author\": \"user\", \"content\": \"Hello what can you do?\"},\n", + " \"task_id\": task.id,\n", + " }\n", + ")\n", + "\n", + "event = rpc_response.result\n", + "print(event)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a6927cc0", + "metadata": {}, + "outputs": [], + "source": [ + "# Subscribe to the async task messages produced by the agent\n", + "from agentex.lib.utils.dev_tools import subscribe_to_async_task_messages\n", + "\n", + "task_messages = subscribe_to_async_task_messages(\n", + " client=client,\n", + " task=task, \n", + " only_after_timestamp=event.created_at, \n", + " print_messages=True,\n", + " rich_print=True,\n", + " timeout=5,\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4864e354", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/environments.yaml.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/environments.yaml.j2 new file mode 100644 index 000000000..f802776f0 --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/environments.yaml.j2 @@ -0,0 +1,57 @@ +# Agent Environment Configuration +# ------------------------------ +# This file defines environment-specific settings for your agent. +# This DIFFERS from the manifest.yaml file in that it is used to program things that are ONLY per environment. + +# ********** EXAMPLE ********** +# schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +# environments: +# dev: +# auth: +# principal: +# user_id: "1234567890" +# user_name: "John Doe" +# user_email: "john.doe@example.com" +# user_role: "admin" +# user_permissions: "read, write, delete" +# helm_overrides: # This is used to override the global helm values.yaml file in the agentex-agent helm charts +# replicas: 3 +# resources: +# requests: +# cpu: "1000m" +# memory: "2Gi" +# limits: +# cpu: "2000m" +# memory: "4Gi" +# env: +# - name: LOG_LEVEL +# value: "DEBUG" +# - name: ENVIRONMENT +# value: "staging" +# +# kubernetes: +# # OPTIONAL - Otherwise it will be derived from separately. However, this can be used to override the derived +# # namespace and deploy it with in the same namespace that already exists for a separate agent. +# namespace: "team-{{agent_name}}" +# ********** END EXAMPLE ********** + +schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +environments: + dev: + auth: + principal: + user_id: # TODO: Fill in + account_id: # TODO: Fill in + helm_overrides: + replicaCount: 2 + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" + temporal: + enabled: false + + diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/manifest.yaml.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/manifest.yaml.j2 new file mode 100644 index 000000000..8928db0ef --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/manifest.yaml.j2 @@ -0,0 +1,123 @@ +# Agent Manifest Configuration +# --------------------------- +# This file defines how your agent should be built and deployed. + +# Build Configuration +# ------------------ +# The build config defines what gets packaged into your agent's Docker image. +# This same configuration is used whether building locally or remotely. +# +# When building: +# 1. All files from include_paths are collected into a build context +# 2. The context is filtered by dockerignore rules +# 3. The Dockerfile uses this context to build your agent's image +# 4. The image is pushed to a registry and used to run your agent +build: + context: + # Root directory for the build context + root: ../ # Keep this as the default root + + # Paths to include in the Docker build context + # Must include: + # - Your agent's directory (your custom agent code) + # These paths are collected and sent to the Docker daemon for building + include_paths: + - {{ project_path_from_build_root }} + + # Path to your agent's Dockerfile + # This defines how your agent's image is built from the context + # Relative to the root directory + dockerfile: {{ project_path_from_build_root }}/Dockerfile + + # Path to your agent's .dockerignore + # Filters unnecessary files from the build context + # Helps keep build context small and builds fast + dockerignore: {{ project_path_from_build_root }}/.dockerignore + + +# Local Development Configuration +# ----------------------------- +# Only used when running the agent locally +local_development: + agent: + port: 8000 # Port where your local ACP server is running + host_address: host.docker.internal # Host address for Docker networking (host.docker.internal for Docker, localhost for direct) + + # File paths for local development (relative to this manifest.yaml) + paths: + # Path to ACP server file + # Examples: + # project/acp.py (standard) + # src/server.py (custom structure) + # ../shared/acp.py (shared across projects) + # /absolute/path/acp.py (absolute path) + acp: project/acp.py + + +# Agent Configuration +# ----------------- +agent: + acp_type: async + + # Unique name for your agent + # Used for task routing and monitoring + name: {{ agent_name }} + + # Description of what your agent does + # Helps with documentation and discovery + description: {{ description | tojson }} + + # Temporal workflow configuration + # Set enabled: true to use Temporal workflows for long-running tasks + temporal: + enabled: false + + # Optional: Credentials mapping + # Maps Kubernetes secrets to environment variables + # Common credentials include: + credentials: + # The Gemini CLI CLI authenticates with GEMINI_API_KEY (LITELLM_API_KEY + # is not read by the `gemini` subprocess this agent spawns). + - env_var_name: GEMINI_API_KEY + secret_name: gemini-api-key + secret_key: api-key + - env_var_name: SGP_API_KEY + secret_name: sgp-api-key + secret_key: api-key + - env_var_name: REDIS_URL + secret_name: redis-url-secret + secret_key: url + + # Optional: Set Environment variables for running your agent locally as well + # as for deployment later on. GEMINI_API_KEY is supplied via the credential + # mapping above (deploy) or your local .env (load_dotenv). Do NOT set it to an + # empty string here — that would shadow the real key at runtime. + env: {} + # GEMINI_API_KEY: "" # uncomment only to hardcode for local runs + +# Deployment Configuration +# ----------------------- +# Configuration for deploying your agent to Kubernetes clusters +deployment: + # Container image configuration + image: + repository: "" # Update with your container registry + tag: "latest" # Default tag, should be versioned in production + + imagePullSecrets: [] # Update with your image pull secret names + # - name: my-registry-secret + + # Global deployment settings that apply to all clusters + # These can be overridden in cluster-specific environments (environments.yaml) + global: + # Default replica count + replicaCount: 1 + + # Default resource requirements + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/project/acp.py.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/project/acp.py.j2 new file mode 100644 index 000000000..6967ec19b --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/project/acp.py.j2 @@ -0,0 +1,167 @@ +"""ACP handler for {{ agent_name }} — an async Gemini CLI agent. + +Spawns ``gemini -p "" --output-format stream-json`` as a LOCAL +asyncio subprocess (no Scale sandbox — that is a production concern). Stdout +lines are fed into ``GeminiCliTurn``. Events are delivered via +``UnifiedEmitter.auto_send_turn``, the async Redis push path. + +Live runs require the ``gemini`` CLI to be installed and an +GEMINI_API_KEY (or equivalent credential) in the environment. +""" + +from __future__ import annotations + +import os +import asyncio +from typing import AsyncIterator +from collections import deque + +from dotenv import load_dotenv + +load_dotenv() + +import agentex.lib.adk as adk +from agentex.lib.adk import GeminiCliTurn +from agentex.lib.types.acp import SendEventParams, CancelTaskParams, CreateTaskParams +from agentex.lib.core.harness import UnifiedEmitter +from agentex.lib.types.fastacp import AsyncACPConfig +from agentex.lib.types.tracing import SGPTracingProcessorConfig +from agentex.lib.utils.logging import make_logger +from agentex.types.text_content import TextContent +from agentex.lib.sdk.fastacp.fastacp import FastACP +from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config + +logger = make_logger(__name__) + +add_tracing_processor_config( + SGPTracingProcessorConfig( + sgp_api_key=os.environ.get("SGP_API_KEY", ""), + sgp_account_id=os.environ.get("SGP_ACCOUNT_ID", ""), + sgp_base_url=os.environ.get("SGP_CLIENT_BASE_URL", ""), + ) +) + +acp = FastACP.create( + acp_type="async", + config=AsyncACPConfig(type="base"), +) + + +async def _spawn_gemini(prompt: str) -> AsyncIterator[str]: + """Spawn ``gemini -p --output-format stream-json`` locally and yield stdout lines. + + Injectable seam: tests can monkeypatch this with a fake async iterator of + pre-recorded lines so no real CLI invocation is needed offline. + """ + # The prompt goes in via ``-p`` (argv). Stdin is closed on purpose: in + # non-interactive mode the CLI reads stdin to EOF and appends it to the + # prompt, so an open pipe would make it wait forever. ``GEMINI_MODEL`` + # (optional) selects the model; the CLI's default is ``auto``. + cmd = ["gemini", "-p", prompt, "--output-format", "stream-json"] + model = os.environ.get("GEMINI_MODEL") + if model: + cmd.extend(["-m", model]) + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert proc.stdout is not None + + # Drain stderr concurrently. The Gemini CLI can write enough to + # stderr to fill the OS pipe buffer; if we only read stdout, the CLI blocks + # on its stderr write while we block reading stdout — a deadlock. A + # background task keeps stderr flowing so stdout never stalls. We keep a + # bounded tail so a non-zero exit can be surfaced with context instead of + # silently completing the turn. + stderr_tail: deque[str] = deque(maxlen=20) + + async def _drain_stderr() -> None: + assert proc.stderr is not None + async for raw in proc.stderr: + text = raw.decode("utf-8", errors="replace").rstrip() + if text: + stderr_tail.append(text) + + stderr_task = asyncio.create_task(_drain_stderr()) + + try: + buffer = "" + async for chunk in proc.stdout: + buffer += chunk.decode("utf-8", errors="replace") + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.strip() + if line: + yield line + + if buffer.strip(): + yield buffer.strip() + + await proc.wait() + if proc.returncode: + # The CLI failed (missing binary/auth, bad command). Raise so the + # turn surfaces as failed instead of completing with no output. + tail = "\n".join(stderr_tail) + raise RuntimeError( + f"gemini CLI exited with status {proc.returncode}:\n{tail}" + ) + finally: + # Release the subprocess and stderr drain task even if the consumer + # abandons the generator early (task cancellation / client disconnect): + # cancel the drain task and terminate+reap the process if it is still + # running, so neither is leaked. + stderr_task.cancel() + try: + await stderr_task + except asyncio.CancelledError: + pass + if proc.returncode is None: + try: + proc.terminate() + except ProcessLookupError: + pass + await proc.wait() + + +@acp.on_task_create +async def handle_task_create(params: CreateTaskParams): + logger.info("Task created: %s", params.task.id) + + +@acp.on_task_event_send +async def handle_task_event_send(params: SendEventParams): + """Handle a user message: spawn Gemini CLI locally and push events to the task stream.""" + task_id = params.task.id + content = params.event.content + if not isinstance(content, TextContent): + logger.warning("Ignoring non-text event content (type=%s)", getattr(content, "type", "?")) + return + prompt = content.content + logger.info("Processing message for task %s", task_id) + + await adk.messages.create(task_id=task_id, content=params.event.content) + + async with adk.tracing.span( + trace_id=task_id, + task_id=task_id, + name="message", + input={"message": prompt}, + data={"__span_type__": "AGENT_WORKFLOW"}, + ) as turn_span: + emitter = UnifiedEmitter( + task_id=task_id, + trace_id=task_id, + parent_span_id=turn_span.id if turn_span else None, + ) + turn = GeminiCliTurn(_spawn_gemini(prompt)) + result = await emitter.auto_send_turn(turn) + if turn_span: + turn_span.output = {"final_text": result.final_text} + + +@acp.on_task_cancel +async def handle_task_canceled(params: CancelTaskParams): + logger.info("Task canceled: %s", params.task.id) diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/pyproject.toml.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/pyproject.toml.j2 new file mode 100644 index 000000000..e499b1dc1 --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/pyproject.toml.j2 @@ -0,0 +1,33 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "{{ project_name }}" +version = "0.1.0" +description = "{{ description }}" +requires-python = ">=3.12" +dependencies = [ + "agentex-sdk", + "scale-gp", + "python-dotenv>=1.0,<2", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "black", + "isort", + "flake8", +] + +[tool.hatch.build.targets.wheel] +packages = ["project"] + +[tool.black] +line-length = 88 +target-version = ['py312'] + +[tool.isort] +profile = "black" +line_length = 88 diff --git a/src/agentex/lib/cli/templates/default-gemini-cli/requirements.txt.j2 b/src/agentex/lib/cli/templates/default-gemini-cli/requirements.txt.j2 new file mode 100644 index 000000000..8c0630384 --- /dev/null +++ b/src/agentex/lib/cli/templates/default-gemini-cli/requirements.txt.j2 @@ -0,0 +1,8 @@ +# Install agentex-sdk from local path +agentex-sdk + +# Scale GenAI Platform Python SDK +scale-gp + +# Loads .env files for local development +python-dotenv>=1.0,<2 diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/.dockerignore.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/.dockerignore.j2 new file mode 100644 index 000000000..c2d7fca4d --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/.dockerignore.j2 @@ -0,0 +1,43 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Environments +.env** +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Git +.git +.gitignore + +# Misc +.DS_Store diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/.env.example.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/.env.example.j2 new file mode 100644 index 000000000..611d7886b --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/.env.example.j2 @@ -0,0 +1,13 @@ +# {{ agent_name }} - Environment Variables +# Copy this file to .env and fill in the values + +# API key for the Gemini CLI (the `gemini` subprocess this agent spawns) +GEMINI_API_KEY= + +# LLM base URL (optional - override to use a different provider) +# OPENAI_BASE_URL= + +# SGP Configuration (optional - for tracing) +# SGP_API_KEY= +# SGP_ACCOUNT_ID= +# SGP_CLIENT_BASE_URL= diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile-uv.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile-uv.j2 new file mode 100644 index 000000000..0f9880ac0 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile-uv.j2 @@ -0,0 +1,51 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/** + +# Install the Gemini CLI CLI: the agent shells out to `gemini` on every turn, +# so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +ENV UV_COMPILE_BYTECODE=1 +ENV UV_LINK_MODE=copy +ENV UV_HTTP_TIMEOUT=1000 + +WORKDIR /app/{{ project_path_from_build_root }} + +# Copy dependency files for layer caching +COPY {{ project_path_from_build_root }}/pyproject.toml ./ + +# Install dependencies (without project itself, for layer caching) +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-install-project --no-dev + +# Copy the project code +COPY {{ project_path_from_build_root }}/project ./project + +# Install the project +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-dev + +ENV PATH="/app/{{ project_path_from_build_root }}/.venv/bin:$PATH" +ENV PYTHONPATH=/app + +# Run the agent using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile.j2 new file mode 100644 index 000000000..7bc746e60 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/Dockerfile.j2 @@ -0,0 +1,47 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/* + +# Install the Gemini CLI CLI: the agent shells out to `gemini` on every turn, +# so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +RUN uv pip install --system --upgrade pip setuptools wheel + +ENV UV_HTTP_TIMEOUT=1000 + +# Copy just the requirements file to optimize caching +COPY {{ project_path_from_build_root }}/requirements.txt /app/{{ project_path_from_build_root }}/requirements.txt + +WORKDIR /app/{{ project_path_from_build_root }} + +# Install the required Python packages +RUN uv pip install --system -r requirements.txt + +# Copy the project code +COPY {{ project_path_from_build_root }}/project /app/{{ project_path_from_build_root }}/project + + +# Set environment variables +ENV PYTHONPATH=/app + +# Run the agent using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/README.md.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/README.md.j2 new file mode 100644 index 000000000..e7aa6737c --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/README.md.j2 @@ -0,0 +1,64 @@ +# {{ agent_name }} - AgentEx Sync Gemini CLI Agent + +This template builds a **synchronous** agent that drives the **Gemini CLI CLI** +through the unified harness surface on AgentEx: +- Spawns `gemini -p "" --output-format stream-json` as a local subprocess +- Wraps the CLI's stdout stream in a `GeminiCliTurn` +- Delivers canonical `StreamTaskMessage*` events via `UnifiedEmitter.yield_turn` + (the sync HTTP yield path) +- Tracing integration to SGP / AgentEx + +## Prerequisites + +- The `gemini` CLI installed and on your `PATH` +- An `GEMINI_API_KEY` (or equivalent credential) in your environment + +## Running the Agent + +```bash +agentex agents run --manifest manifest.yaml +``` + +## Project Structure + +``` +{{ project_name }}/ +├── project/ +│ ├── __init__.py +│ └── acp.py # ACP server, subprocess spawn, and message handler +├── Dockerfile +├── manifest.yaml +├── dev.ipynb +{% if use_uv %} +└── pyproject.toml +{% else %} +└── requirements.txt +{% endif %} +``` + +## Key Concepts + +### Sync ACP with the harness +The sync ACP model uses HTTP request/response. The `@acp.on_message_send` +handler spawns the Gemini CLI CLI and yields the harness events back to the +client as they arrive. + +### The unified harness surface +`GeminiCliTurn` + `UnifiedEmitter` are the unified harness surface. The turn +normalizes CLI output into canonical AgentEx events; the emitter traces and +delivers them. + +## Development + +### 1. Customize the subprocess +Edit `_spawn_gemini` in `project/acp.py` to change the CLI flags, working +directory, or how the prompt is delivered. + +### 2. Configure Credentials +Set your credentials via `manifest.yaml`, an exported environment variable, or a +`.env` file in the project directory. + +### 3. Run Locally +```bash +export ENVIRONMENT=development && agentex agents run --manifest manifest.yaml +``` diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/dev.ipynb.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/dev.ipynb.j2 new file mode 100644 index 000000000..b0691b1b1 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/dev.ipynb.j2 @@ -0,0 +1,167 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "36834357", + "metadata": {}, + "outputs": [], + "source": [ + "from agentex import Agentex\n", + "\n", + "client = Agentex(base_url=\"http://localhost:5003\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d1c309d6", + "metadata": {}, + "outputs": [], + "source": [ + "AGENT_NAME = \"{{ agent_name }}\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9f6e6ef0", + "metadata": {}, + "outputs": [], + "source": [ + "# # (Optional) Create a new task. If you don't create a new task, each message will be sent to a new task. The server will create the task for you.\n", + "\n", + "# import uuid\n", + "\n", + "# TASK_ID = str(uuid.uuid4())[:8]\n", + "\n", + "# rpc_response = client.agents.rpc_by_name(\n", + "# agent_name=AGENT_NAME,\n", + "# method=\"task/create\",\n", + "# params={\n", + "# \"name\": f\"{TASK_ID}-task\",\n", + "# \"params\": {}\n", + "# }\n", + "# )\n", + "\n", + "# task = rpc_response.result\n", + "# print(task)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b03b0d37", + "metadata": {}, + "outputs": [], + "source": [ + "# Test non streaming response\n", + "from agentex.types import TextContent\n", + "\n", + "# The response is expected to be a list of TaskMessage objects, which is a union of the following types:\n", + "# - TextContent: A message with just text content \n", + "# - DataContent: A message with JSON-serializable data content\n", + "# - ToolRequestContent: A message with a tool request, which contains a JSON-serializable request to call a tool\n", + "# - ToolResponseContent: A message with a tool response, which contains response object from a tool call in its content\n", + "\n", + "# When processing the message/send response, if you are expecting more than TextContent, such as DataContent, ToolRequestContent, or ToolResponseContent, you can process them as well\n", + "\n", + "rpc_response = client.agents.send_message(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"content\": {\"type\": \"text\", \"author\": \"user\", \"content\": \"Hello what can you do?\"},\n", + " \"stream\": False\n", + " }\n", + ")\n", + "\n", + "if not rpc_response or not rpc_response.result:\n", + " raise ValueError(\"No result in response\")\n", + "\n", + "# Extract and print just the text content from the response\n", + "for task_message in rpc_response.result:\n", + " content = task_message.content\n", + " if isinstance(content, TextContent):\n", + " text = content.content\n", + " print(text)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "79688331", + "metadata": {}, + "outputs": [], + "source": [ + "# Test streaming response\n", + "from agentex.types.task_message_update import StreamTaskMessageDelta, StreamTaskMessageFull\n", + "from agentex.types.text_delta import TextDelta\n", + "\n", + "\n", + "# The result object of message/send will be a TaskMessageUpdate which is a union of the following types:\n", + "# - StreamTaskMessageStart: \n", + "# - An indicator that a streaming message was started, doesn't contain any useful content\n", + "# - StreamTaskMessageDelta: \n", + "# - A delta of a streaming message, contains the text delta to aggregate\n", + "# - StreamTaskMessageDone: \n", + "# - An indicator that a streaming message was done, doesn't contain any useful content\n", + "# - StreamTaskMessageFull: \n", + "# - A non-streaming message, there is nothing to aggregate, since this contains the full message, not deltas\n", + "\n", + "# Whenn processing StreamTaskMessageDelta, if you are expecting more than TextDeltas, such as DataDelta, ToolRequestDelta, or ToolResponseDelta, you can process them as well\n", + "# Whenn processing StreamTaskMessageFull, if you are expecting more than TextContent, such as DataContent, ToolRequestContent, or ToolResponseContent, you can process them as well\n", + "\n", + "for agent_rpc_response_chunk in client.agents.send_message_stream(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"content\": {\"type\": \"text\", \"author\": \"user\", \"content\": \"Hello what can you do?\"},\n", + " \"stream\": True\n", + " }\n", + "):\n", + " # We know that the result of the message/send when stream is set to True will be a TaskMessageUpdate\n", + " task_message_update = agent_rpc_response_chunk.result\n", + " # Print oly the text deltas as they arrive or any full messages\n", + " if isinstance(task_message_update, StreamTaskMessageDelta):\n", + " delta = task_message_update.delta\n", + " if isinstance(delta, TextDelta):\n", + " print(delta.text_delta, end=\"\", flush=True)\n", + " else:\n", + " print(f\"Found non-text {type(task_message_update)} object in streaming message.\")\n", + " elif isinstance(task_message_update, StreamTaskMessageFull):\n", + " content = task_message_update.content\n", + " if isinstance(content, TextContent):\n", + " print(content.content)\n", + " else:\n", + " print(f\"Found non-text {type(task_message_update)} object in full message.\")\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c5e7e042", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/environments.yaml.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/environments.yaml.j2 new file mode 100644 index 000000000..73924abdd --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/environments.yaml.j2 @@ -0,0 +1,53 @@ +# Agent Environment Configuration +# ------------------------------ +# This file defines environment-specific settings for your agent. +# This DIFFERS from the manifest.yaml file in that it is used to program things that are ONLY per environment. + +# ********** EXAMPLE ********** +# schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +# environments: +# dev: +# auth: +# principal: +# user_id: "1234567890" +# user_name: "John Doe" +# user_email: "john.doe@example.com" +# user_role: "admin" +# user_permissions: "read, write, delete" +# helm_overrides: # This is used to override the global helm values.yaml file in the agentex-agent helm charts +# replicas: 3 +# resources: +# requests: +# cpu: "1000m" +# memory: "2Gi" +# limits: +# cpu: "2000m" +# memory: "4Gi" +# env: +# - name: LOG_LEVEL +# value: "DEBUG" +# - name: ENVIRONMENT +# value: "staging" +# kubernetes: +# # OPTIONAL - Otherwise it will be derived from separately. However, this can be used to override the derived +# # namespace and deploy it with in the same namespace that already exists for a separate agent. +# namespace: "team-{{agent_name}}" +# ********** END EXAMPLE ********** + +schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +environments: + dev: + auth: + principal: + user_id: # TODO: Fill in + account_id: # TODO: Fill in + helm_overrides: + replicaCount: 2 + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" + diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/manifest.yaml.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/manifest.yaml.j2 new file mode 100644 index 000000000..758b0abd5 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/manifest.yaml.j2 @@ -0,0 +1,120 @@ +# Agent Manifest Configuration +# --------------------------- +# This file defines how your agent should be built and deployed. + +# Build Configuration +# ------------------ +# The build config defines what gets packaged into your agent's Docker image. +# This same configuration is used whether building locally or remotely. +# +# When building: +# 1. All files from include_paths are collected into a build context +# 2. The context is filtered by dockerignore rules +# 3. The Dockerfile uses this context to build your agent's image +# 4. The image is pushed to a registry and used to run your agent +build: + context: + # Root directory for the build context + root: ../ # Keep this as the default root + + # Paths to include in the Docker build context + # Must include: + # - Your agent's directory (your custom agent code) + # These paths are collected and sent to the Docker daemon for building + include_paths: + - {{ project_path_from_build_root }} + + # Path to your agent's Dockerfile + # This defines how your agent's image is built from the context + # Relative to the root directory + dockerfile: {{ project_path_from_build_root }}/Dockerfile + + # Path to your agent's .dockerignore + # Filters unnecessary files from the build context + # Helps keep build context small and builds fast + dockerignore: {{ project_path_from_build_root }}/.dockerignore + + +# Local Development Configuration +# ----------------------------- +# Only used when running the agent locally +local_development: + agent: + port: 8000 # Port where your local ACP server is running + host_address: host.docker.internal # Host address for Docker networking (host.docker.internal for Docker, localhost for direct) + + # File paths for local development (relative to this manifest.yaml) + paths: + # Path to ACP server file + # Examples: + # project/acp.py (standard) + # src/server.py (custom structure) + # ../shared/acp.py (shared across projects) + # /absolute/path/acp.py (absolute path) + acp: project/acp.py + + +# Agent Configuration +# ----------------- +agent: + acp_type: sync + # Unique name for your agent + # Used for task routing and monitoring + name: {{ agent_name }} + + # Description of what your agent does + # Helps with documentation and discovery + description: {{ description | tojson }} + + # Temporal workflow configuration + # Set enabled: true to use Temporal workflows for long-running tasks + temporal: + enabled: false + + # Optional: Credentials mapping + # Maps Kubernetes secrets to environment variables + # Common credentials include: + credentials: + # The Gemini CLI CLI authenticates with GEMINI_API_KEY (LITELLM_API_KEY + # is not read by the `gemini` subprocess this agent spawns). + - env_var_name: GEMINI_API_KEY + secret_name: gemini-api-key + secret_key: api-key + - env_var_name: SGP_API_KEY + secret_name: sgp-api-key + secret_key: api-key + + # Optional: Set Environment variables for running your agent locally as well + # as for deployment later on. GEMINI_API_KEY is supplied via the credential + # mapping above (deploy) or your local .env (load_dotenv). Do NOT set it to an + # empty string here — that would shadow the real key at runtime. + env: {} + # GEMINI_API_KEY: "" # uncomment only to hardcode for local runs + + +# Deployment Configuration +# ----------------------- +# Configuration for deploying your agent to Kubernetes clusters +deployment: + # Container image configuration + image: + repository: "" # Update with your container registry + tag: "latest" # Default tag, should be versioned in production + + imagePullSecrets: [] # Update with your image pull secret names + # - name: my-registry-secret + + # Global deployment settings that apply to all clusters + # These can be overridden in cluster-specific environments (environments.yaml) + global: + # Default replica count + replicaCount: 1 + + # Default resource requirements + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/project/acp.py.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/project/acp.py.j2 new file mode 100644 index 000000000..73c35a735 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/project/acp.py.j2 @@ -0,0 +1,155 @@ +"""ACP handler for {{ agent_name }} — a sync Gemini CLI agent. + +Spawns ``gemini -p "" --output-format stream-json`` as a LOCAL +asyncio subprocess (no Scale sandbox — that is a production concern). Stdout +lines are fed into ``GeminiCliTurn``, which wraps +``convert_gemini_cli_to_agentex_events``. Events are delivered via +``UnifiedEmitter.yield_turn``, the sync HTTP yield path. + +Live runs require the ``gemini`` CLI to be installed and an +GEMINI_API_KEY (or equivalent credential) to be in the environment. +""" + +from __future__ import annotations + +import os +import asyncio +from typing import AsyncIterator, AsyncGenerator +from collections import deque + +from dotenv import load_dotenv + +load_dotenv() + +import agentex.lib.adk as adk +from agentex.lib.adk import GeminiCliTurn +from agentex.lib.types.acp import SendMessageParams +from agentex.lib.core.harness import UnifiedEmitter +from agentex.lib.types.tracing import SGPTracingProcessorConfig +from agentex.lib.utils.logging import make_logger +from agentex.types.text_content import TextContent +from agentex.lib.sdk.fastacp.fastacp import FastACP +from agentex.types.task_message_update import TaskMessageUpdate +from agentex.types.task_message_content import TaskMessageContent +from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config + +logger = make_logger(__name__) + +add_tracing_processor_config( + SGPTracingProcessorConfig( + sgp_api_key=os.environ.get("SGP_API_KEY", ""), + sgp_account_id=os.environ.get("SGP_ACCOUNT_ID", ""), + sgp_base_url=os.environ.get("SGP_CLIENT_BASE_URL", ""), + ) +) + +acp = FastACP.create(acp_type="sync") + + +async def _spawn_gemini(prompt: str) -> AsyncIterator[str]: + """Spawn ``gemini -p --output-format stream-json`` locally and yield stdout lines. + + This is a seam: tests can replace it with a fake async iterator of + pre-recorded lines so no real CLI invocation is needed offline. + """ + # The prompt goes in via ``-p`` (argv). Stdin is closed on purpose: in + # non-interactive mode the CLI reads stdin to EOF and appends it to the + # prompt, so an open pipe would make it wait forever. ``GEMINI_MODEL`` + # (optional) selects the model; the CLI's default is ``auto``. + cmd = ["gemini", "-p", prompt, "--output-format", "stream-json"] + model = os.environ.get("GEMINI_MODEL") + if model: + cmd.extend(["-m", model]) + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert proc.stdout is not None + + # Drain stderr concurrently. The Gemini CLI can write enough to + # stderr to fill the OS pipe buffer; if we only read stdout, the CLI blocks + # on its stderr write while we block reading stdout — a deadlock. A + # background task keeps stderr flowing so stdout never stalls. We keep a + # bounded tail so a non-zero exit can be surfaced with context instead of + # silently completing the turn. + stderr_tail: deque[str] = deque(maxlen=20) + + async def _drain_stderr() -> None: + assert proc.stderr is not None + async for raw in proc.stderr: + text = raw.decode("utf-8", errors="replace").rstrip() + if text: + stderr_tail.append(text) + + stderr_task = asyncio.create_task(_drain_stderr()) + + try: + buffer = "" + async for chunk in proc.stdout: + buffer += chunk.decode("utf-8", errors="replace") + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.strip() + if line: + yield line + + if buffer.strip(): + yield buffer.strip() + + await proc.wait() + if proc.returncode: + # The CLI failed (missing binary/auth, bad command). Raise so the + # turn surfaces as failed instead of completing with no output. + tail = "\n".join(stderr_tail) + raise RuntimeError( + f"gemini CLI exited with status {proc.returncode}:\n{tail}" + ) + finally: + # Release the subprocess and stderr drain task even if the consumer + # abandons the generator early (task cancellation / client disconnect): + # cancel the drain task and terminate+reap the process if it is still + # running, so neither is leaked. + stderr_task.cancel() + try: + await stderr_task + except asyncio.CancelledError: + pass + if proc.returncode is None: + try: + proc.terminate() + except ProcessLookupError: + pass + await proc.wait() + + +@acp.on_message_send +async def handle_message_send( + params: SendMessageParams, +) -> TaskMessageContent | list[TaskMessageContent] | AsyncGenerator[TaskMessageUpdate, None]: + """Handle an incoming message: run Gemini CLI locally and stream events.""" + task_id = params.task.id + content = params.content + if not isinstance(content, TextContent): + logger.warning("Ignoring non-text message content (type=%s)", getattr(content, "type", "?")) + return + prompt = content.content + logger.info("Processing message for task %s", task_id) + + async with adk.tracing.span( + trace_id=task_id, + task_id=task_id, + name="message", + input={"message": prompt}, + data={"__span_type__": "AGENT_WORKFLOW"}, + ) as turn_span: + emitter = UnifiedEmitter( + task_id=task_id, + trace_id=task_id, + parent_span_id=turn_span.id if turn_span else None, + ) + turn = GeminiCliTurn(_spawn_gemini(prompt)) + async for event in emitter.yield_turn(turn): + yield event diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/pyproject.toml.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/pyproject.toml.j2 new file mode 100644 index 000000000..e499b1dc1 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/pyproject.toml.j2 @@ -0,0 +1,33 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "{{ project_name }}" +version = "0.1.0" +description = "{{ description }}" +requires-python = ">=3.12" +dependencies = [ + "agentex-sdk", + "scale-gp", + "python-dotenv>=1.0,<2", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "black", + "isort", + "flake8", +] + +[tool.hatch.build.targets.wheel] +packages = ["project"] + +[tool.black] +line-length = 88 +target-version = ['py312'] + +[tool.isort] +profile = "black" +line_length = 88 diff --git a/src/agentex/lib/cli/templates/sync-gemini-cli/requirements.txt.j2 b/src/agentex/lib/cli/templates/sync-gemini-cli/requirements.txt.j2 new file mode 100644 index 000000000..8c0630384 --- /dev/null +++ b/src/agentex/lib/cli/templates/sync-gemini-cli/requirements.txt.j2 @@ -0,0 +1,8 @@ +# Install agentex-sdk from local path +agentex-sdk + +# Scale GenAI Platform Python SDK +scale-gp + +# Loads .env files for local development +python-dotenv>=1.0,<2 diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/.dockerignore.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/.dockerignore.j2 new file mode 100644 index 000000000..c2d7fca4d --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/.dockerignore.j2 @@ -0,0 +1,43 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Environments +.env** +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Git +.git +.gitignore + +# Misc +.DS_Store diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/.env.example.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/.env.example.j2 new file mode 100644 index 000000000..611d7886b --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/.env.example.j2 @@ -0,0 +1,13 @@ +# {{ agent_name }} - Environment Variables +# Copy this file to .env and fill in the values + +# API key for the Gemini CLI (the `gemini` subprocess this agent spawns) +GEMINI_API_KEY= + +# LLM base URL (optional - override to use a different provider) +# OPENAI_BASE_URL= + +# SGP Configuration (optional - for tracing) +# SGP_API_KEY= +# SGP_ACCOUNT_ID= +# SGP_CLIENT_BASE_URL= diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile-uv.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile-uv.j2 new file mode 100644 index 000000000..ea4a58f82 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile-uv.j2 @@ -0,0 +1,61 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/** + +# Install the Gemini CLI CLI: the activity shells out to `gemini` on every +# turn, so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +# Install tctl (Temporal CLI) +RUN ARCH="$(uname -m)" && \ + case "$ARCH" in x86_64) TCTL_ARCH=amd64 ;; aarch64|arm64) TCTL_ARCH=arm64 ;; *) TCTL_ARCH=amd64 ;; esac && \ + curl -L "https://github.com/temporalio/tctl/releases/download/v1.18.1/tctl_1.18.1_linux_${TCTL_ARCH}.tar.gz" -o /tmp/tctl.tar.gz && \ + tar -xzf /tmp/tctl.tar.gz -C /usr/local/bin && \ + chmod +x /usr/local/bin/tctl && \ + rm /tmp/tctl.tar.gz + +ENV UV_COMPILE_BYTECODE=1 +ENV UV_LINK_MODE=copy +ENV UV_HTTP_TIMEOUT=1000 + +WORKDIR /app/{{ project_path_from_build_root }} + +# Copy dependency files for layer caching +COPY {{ project_path_from_build_root }}/pyproject.toml ./ + +# Install dependencies (without project itself, for layer caching) +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-install-project --no-dev + +# Copy the project code +COPY {{ project_path_from_build_root }}/project ./project + +# Install the project +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --no-dev + +ENV PATH="/app/{{ project_path_from_build_root }}/.venv/bin:$PATH" + +# Run the ACP server using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] + +# When we deploy the worker, we will replace the CMD with the following +# CMD ["python", "-m", "run_worker"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile.j2 new file mode 100644 index 000000000..9b5e38fa5 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/Dockerfile.j2 @@ -0,0 +1,54 @@ +# syntax=docker/dockerfile:1.3 +FROM python:3.12-slim +COPY --from=ghcr.io/astral-sh/uv:0.6.4 /uv /uvx /bin/ + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + htop \ + vim \ + curl \ + tar \ + python3-dev \ + postgresql-client \ + build-essential \ + libpq-dev \ + gcc \ + cmake \ + netcat-openbsd \ + nodejs \ + npm \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/* + +# Install the Gemini CLI CLI: the activity shells out to `gemini` on every +# turn, so the binary must be present in the runtime image. +RUN npm install -g @google/gemini-cli + +# Install tctl (Temporal CLI) +RUN ARCH="$(uname -m)" && \ + case "$ARCH" in x86_64) TCTL_ARCH=amd64 ;; aarch64|arm64) TCTL_ARCH=arm64 ;; *) TCTL_ARCH=amd64 ;; esac && \ + curl -L "https://github.com/temporalio/tctl/releases/download/v1.18.1/tctl_1.18.1_linux_${TCTL_ARCH}.tar.gz" -o /tmp/tctl.tar.gz && \ + tar -xzf /tmp/tctl.tar.gz -C /usr/local/bin && \ + chmod +x /usr/local/bin/tctl && \ + rm /tmp/tctl.tar.gz + +RUN uv pip install --system --upgrade pip setuptools wheel + +ENV UV_HTTP_TIMEOUT=1000 + +# Copy just the requirements file to optimize caching +COPY {{ project_path_from_build_root }}/requirements.txt /app/{{ project_path_from_build_root }}/requirements.txt + +WORKDIR /app/{{ project_path_from_build_root }} + +# Install the required Python packages +RUN uv pip install --system -r requirements.txt + +# Copy the project code +COPY {{ project_path_from_build_root }}/project /app/{{ project_path_from_build_root }}/project + +# Run the ACP server using uvicorn +CMD ["uvicorn", "project.acp:acp", "--host", "0.0.0.0", "--port", "8000"] + +# When we deploy the worker, we will replace the CMD with the following +# CMD ["python", "-m", "run_worker"] \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/README.md.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/README.md.j2 new file mode 100644 index 000000000..9eba1bf32 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/README.md.j2 @@ -0,0 +1,72 @@ +# {{ agent_name }} — AgentEx Temporal + Gemini CLI + +This template builds a **Temporal-durable** agent that drives the **Gemini CLI +CLI** through the unified harness surface on AgentEx: +- A Temporal workflow holds per-task state durably across worker crashes; each turn runs the CLI as an independent prompt (the Gemini CLI does not resume a session by id in headless mode) +- Each turn delegates to the `run_gemini_cli_turn` activity, which spawns the + CLI (subprocess I/O is not permitted on the workflow event loop) +- The activity wraps the CLI's stdout stream in a `GeminiCliTurn` and delivers + canonical `StreamTaskMessage*` events via `UnifiedEmitter.auto_send_turn` +- Tracing integration to SGP / AgentEx + +## Prerequisites + +- The `gemini` CLI installed and on your `PATH` +- An `GEMINI_API_KEY` (or equivalent credential) in your environment +- A running Temporal service (provided automatically by the local dev stack) + +## Running the Agent + +```bash +agentex agents run --manifest manifest.yaml +``` + +This starts both the ACP HTTP server and the Temporal worker. + +## Project Structure + +``` +{{ project_name }}/ +├── project/ +│ ├── __init__.py +│ ├── acp.py # Thin ACP server; FastACP auto-wires to the workflow +│ ├── workflow.py # Temporal workflow (durable conversation state) +│ ├── activities.py # run_gemini_cli_turn activity (CLI subprocess) +│ └── run_worker.py # Temporal worker entrypoint +├── Dockerfile +├── manifest.yaml +├── dev.ipynb +{% if use_uv %} +└── pyproject.toml +{% else %} +└── requirements.txt +{% endif %} +``` + +## Key Concepts + +### Subprocess must run in an activity +Temporal runs workflow + signal-handler bodies on a deterministic sandbox event +loop that does not implement `subprocess_exec`. The workflow therefore delegates +each turn to the `run_gemini_cli_turn` activity, which also gains Temporal's +retry + timeout guarantees. + +### The unified harness surface +`GeminiCliTurn` + `UnifiedEmitter` are the unified harness surface. The turn +normalizes CLI output into canonical AgentEx events; the emitter traces and +delivers them. + +## Development + +### 1. Customize the subprocess +Edit `_spawn_gemini` in `project/activities.py` to change the CLI flags, working +directory, or how the prompt is delivered. + +### 2. Configure Credentials +Set your credentials via `manifest.yaml`, an exported environment variable, or a +`.env` file in the project directory. + +### 3. Run Locally +```bash +export ENVIRONMENT=development && agentex agents run --manifest manifest.yaml +``` diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/dev.ipynb.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/dev.ipynb.j2 new file mode 100644 index 000000000..d3a68303f --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/dev.ipynb.j2 @@ -0,0 +1,126 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "36834357", + "metadata": {}, + "outputs": [], + "source": [ + "from agentex import Agentex\n", + "\n", + "client = Agentex(base_url=\"http://localhost:5003\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d1c309d6", + "metadata": {}, + "outputs": [], + "source": [ + "AGENT_NAME = \"{{ agent_name }}\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9f6e6ef0", + "metadata": {}, + "outputs": [], + "source": [ + "# (REQUIRED) Create a new task. For Async agents, you must create a task for messages to be associated with.\n", + "import uuid\n", + "\n", + "rpc_response = client.agents.create_task(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"name\": f\"{str(uuid.uuid4())[:8]}-task\",\n", + " \"params\": {}\n", + " }\n", + ")\n", + "\n", + "task = rpc_response.result\n", + "print(task)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b03b0d37", + "metadata": {}, + "outputs": [], + "source": [ + "# Send an event to the agent\n", + "\n", + "# The response is expected to be a list of TaskMessage objects, which is a union of the following types:\n", + "# - TextContent: A message with just text content \n", + "# - DataContent: A message with JSON-serializable data content\n", + "# - ToolRequestContent: A message with a tool request, which contains a JSON-serializable request to call a tool\n", + "# - ToolResponseContent: A message with a tool response, which contains response object from a tool call in its content\n", + "\n", + "# When processing the message/send response, if you are expecting more than TextContent, such as DataContent, ToolRequestContent, or ToolResponseContent, you can process them as well\n", + "\n", + "rpc_response = client.agents.send_event(\n", + " agent_name=AGENT_NAME,\n", + " params={\n", + " \"content\": {\"type\": \"text\", \"author\": \"user\", \"content\": \"Hello what can you do?\"},\n", + " \"task_id\": task.id,\n", + " }\n", + ")\n", + "\n", + "event = rpc_response.result\n", + "print(event)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a6927cc0", + "metadata": {}, + "outputs": [], + "source": [ + "# Subscribe to the async task messages produced by the agent\n", + "from agentex.lib.utils.dev_tools import subscribe_to_async_task_messages\n", + "\n", + "task_messages = subscribe_to_async_task_messages(\n", + " client=client,\n", + " task=task, \n", + " only_after_timestamp=event.created_at, \n", + " print_messages=True,\n", + " rich_print=True,\n", + " timeout=5,\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "4864e354", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/environments.yaml.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/environments.yaml.j2 new file mode 100644 index 000000000..a3df5e228 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/environments.yaml.j2 @@ -0,0 +1,64 @@ +# Agent Environment Configuration +# ------------------------------ +# This file defines environment-specific settings for your agent. +# This DIFFERS from the manifest.yaml file in that it is used to program things that are ONLY per environment. + +# ********** EXAMPLE ********** +# schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +# environments: +# dev: +# auth: +# principal: +# user_id: "1234567890" +# user_name: "John Doe" +# user_email: "john.doe@example.com" +# user_role: "admin" +# user_permissions: "read, write, delete" +# helm_overrides: # This is used to override the global helm values.yaml file in the agentex-agent helm charts +# replicas: 3 +# resources: +# requests: +# cpu: "1000m" +# memory: "2Gi" +# limits: +# cpu: "2000m" +# memory: "4Gi" +# env: +# - name: LOG_LEVEL +# value: "DEBUG" +# - name: ENVIRONMENT +# value: "staging" +# +# kubernetes: +# # OPTIONAL - Otherwise it will be derived from separately. However, this can be used to override the derived +# # namespace and deploy it with in the same namespace that already exists for a separate agent. +# namespace: "team-{{agent_name}}" +# ********** END EXAMPLE ********** + +schema_version: "v1" # This is used to validate the file structure and is not used by the agentex CLI +environments: + dev: + auth: + principal: + user_id: # TODO: Fill in + account_id: # TODO: Fill in + helm_overrides: + # This is used to override the global helm values.yaml file in the agentex-agent helm charts + replicaCount: 2 + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" + temporal-worker: + enabled: true + replicaCount: 2 + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/manifest.yaml.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/manifest.yaml.j2 new file mode 100644 index 000000000..7c7a10bc2 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/manifest.yaml.j2 @@ -0,0 +1,142 @@ +# Agent Manifest Configuration +# --------------------------- +# This file defines how your agent should be built and deployed. + +# Build Configuration +# ------------------ +# The build config defines what gets packaged into your agent's Docker image. +# This same configuration is used whether building locally or remotely. +# +# When building: +# 1. All files from include_paths are collected into a build context +# 2. The context is filtered by dockerignore rules +# 3. The Dockerfile uses this context to build your agent's image +# 4. The image is pushed to a registry and used to run your agent +build: + context: + # Root directory for the build context + root: ../ # Keep this as the default root + + # Paths to include in the Docker build context + # Must include: + # - Your agent's directory (your custom agent code) + # These paths are collected and sent to the Docker daemon for building + include_paths: + - {{ project_path_from_build_root }} + + # Path to your agent's Dockerfile + # This defines how your agent's image is built from the context + # Relative to the root directory + dockerfile: {{ project_path_from_build_root }}/Dockerfile + + # Path to your agent's .dockerignore + # Filters unnecessary files from the build context + # Helps keep build context small and builds fast + dockerignore: {{ project_path_from_build_root }}/.dockerignore + + +# Local Development Configuration +# ----------------------------- +# Only used when running the agent locally +local_development: + agent: + port: 8000 # Port where your local ACP server is running + host_address: host.docker.internal # Host address for Docker networking (host.docker.internal for Docker, localhost for direct) + + # File paths for local development (relative to this manifest.yaml) + paths: + # Path to ACP server file + # Examples: + # project/acp.py (standard) + # src/server.py (custom structure) + # ../shared/acp.py (shared across projects) + # /absolute/path/acp.py (absolute path) + acp: project/acp.py + + # Path to temporal worker file + # Examples: + # project/run_worker.py (standard) + # workers/temporal.py (custom structure) + # ../shared/worker.py (shared across projects) + worker: project/run_worker.py + + +# Agent Configuration +# ----------------- +agent: + # Type of agent - either sync or async + acp_type: async + + # Unique name for your agent + # Used for task routing and monitoring + name: {{ agent_name }} + + # Description of what your agent does + # Helps with documentation and discovery + description: {{ description | tojson }} + + # Temporal workflow configuration + # This enables your agent to run as a Temporal workflow for long-running tasks + temporal: + enabled: true + workflows: + # Name of the workflow class + # Must match the @workflow.defn name in your workflow.py + - name: {{ workflow_name }} + + # Queue name for task distribution + # Used by Temporal to route tasks to your agent + # Convention: _task_queue + queue_name: {{ queue_name }} + + # Optional: Health check port for temporal worker + # Defaults to 80 if not specified + # health_check_port: 80 + + # Optional: Credentials mapping + # Maps Kubernetes secrets to environment variables + # Common credentials include: + credentials: + - env_var_name: REDIS_URL + secret_name: redis-url-secret + secret_key: url + # The Gemini CLI CLI spawned in project/activities.py authenticates with + # GEMINI_API_KEY; without it every turn fails with a CLI auth error. + - env_var_name: GEMINI_API_KEY + secret_name: gemini-api-key + secret_key: api-key + + # Optional: Set Environment variables for running your agent locally as well + # as for deployment later on. GEMINI_API_KEY is supplied via the credential + # mapping above (deploy) or your local .env (load_dotenv). Do NOT set it to an + # empty string here — that would shadow the real key at runtime. + env: {} + # GEMINI_API_KEY: "" # uncomment only to hardcode for local runs + + +# Deployment Configuration +# ----------------------- +# Configuration for deploying your agent to Kubernetes clusters +deployment: + # Container image configuration + image: + repository: "" # Update with your container registry + tag: "latest" # Default tag, should be versioned in production + + imagePullSecrets: [] # Update with your image pull secret name + # - name: my-registry-secret + + # Global deployment settings that apply to all clusters + # These can be overridden in cluster-specific environments (environments.yaml) + global: + # Default replica count + replicaCount: 1 + + # Default resource requirements + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "1000m" + memory: "2Gi" \ No newline at end of file diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/project/acp.py.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/acp.py.j2 new file mode 100644 index 000000000..fb77bd3a1 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/acp.py.j2 @@ -0,0 +1,31 @@ +"""ACP server for {{ agent_name }} — a Temporal Gemini CLI agent. + +This file is intentionally thin. When ``acp_type="async"`` is combined +with ``TemporalACPConfig``, FastACP auto-wires: + + HTTP task/create -> @workflow.run on the workflow class + HTTP task/event/send -> @workflow.signal(SignalName.RECEIVE_EVENT) + HTTP task/cancel -> workflow cancellation via the Temporal client + +The actual agent code lives in ``project/workflow.py`` and is executed by +the Temporal worker (``project/run_worker.py``), not by this HTTP process. +""" + +from __future__ import annotations + +import os + +from dotenv import load_dotenv + +load_dotenv() + +from agentex.lib.types.fastacp import TemporalACPConfig +from agentex.lib.sdk.fastacp.fastacp import FastACP + +acp = FastACP.create( + acp_type="async", + config=TemporalACPConfig( + type="temporal", + temporal_address=os.getenv("TEMPORAL_ADDRESS", "localhost:7233"), + ), +) diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/project/activities.py.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/activities.py.j2 new file mode 100644 index 000000000..a7331600e --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/activities.py.j2 @@ -0,0 +1,156 @@ +"""Temporal activity for {{ agent_name }} — Gemini CLI harness. + +Subprocess spawning (and any other I/O) must run inside a Temporal *activity*, +not in workflow code. Temporal runs workflow + signal-handler bodies on a +deterministic sandbox event loop that does not implement ``subprocess_exec`` +(or threads / sockets), so spawning the CLI directly in the signal handler +raises ``NotImplementedError``. This activity runs the Gemini CLI CLI, drives +the ``GeminiCliTurn`` through ``UnifiedEmitter.auto_send_turn`` (the async +Redis push path), and returns the turn result to the workflow. + +The ``_spawn_gemini`` async generator is an injectable seam: offline tests +can provide a fake that yields pre-recorded stdout lines so no real CLI runs. +""" + +from __future__ import annotations + +import os +import asyncio +from typing import Any, AsyncIterator +from datetime import datetime +from collections import deque + +from temporalio import activity + +from agentex.lib.adk import GeminiCliTurn +from agentex.lib.core.harness import UnifiedEmitter +from agentex.lib.utils.logging import make_logger +from agentex.lib.utils.model_utils import BaseModel + +logger = make_logger(__name__) + +RUN_GEMINI_CLI_TURN_ACTIVITY = "run_gemini_cli_turn" + + +class RunGeminiCliTurnParams(BaseModel): + """Arguments for one Gemini CLI turn run inside an activity.""" + + task_id: str + prompt: str + trace_id: str | None = None + parent_span_id: str | None = None + session_id: str | None = None + created_at: datetime | None = None + + +class RunGeminiCliTurnResult(BaseModel): + """Result returned from the activity to the workflow.""" + + final_text: str + session_id: str | None = None + + +async def _spawn_gemini(prompt: str, session_id: str | None = None) -> AsyncIterator[str]: + """Spawn ``gemini -p --output-format stream-json`` locally and yield stdout lines. + + ``session_id`` is accepted for parity with the other CLI harnesses; see the + note in the body about why it is not used for resume. + + Injectable seam: tests can monkeypatch this with a fake async iterator so no + real CLI invocation is needed offline. + """ + # The prompt goes in via ``-p`` (argv). Stdin is closed on purpose: in + # non-interactive mode the CLI reads stdin to EOF and appends it to the + # prompt, so an open pipe would make it wait forever. ``GEMINI_MODEL`` + # (optional) selects the model; the CLI's default is ``auto``. + # + # ``session_id`` is accepted for parity with the other CLI harnesses but not + # used: the Gemini CLI's ``--resume`` takes "latest" or an index, not a + # session id, which is not safe when a worker serves several tasks. Each + # turn therefore runs as an independent prompt. + del session_id + cmd = ["gemini", "-p", prompt, "--output-format", "stream-json"] + model = os.environ.get("GEMINI_MODEL") + if model: + cmd.extend(["-m", model]) + + proc = await asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert proc.stdout is not None + + # Drain stderr concurrently. The Gemini CLI can write enough to + # stderr to fill the OS pipe buffer; if we only read stdout, the CLI blocks + # on its stderr write while we block reading stdout — a deadlock. A + # background task keeps stderr flowing so stdout never stalls. We keep a + # bounded tail so a non-zero exit can be surfaced with context instead of + # silently completing the turn. + stderr_tail: deque[str] = deque(maxlen=20) + + async def _drain_stderr() -> None: + assert proc.stderr is not None + async for raw in proc.stderr: + text = raw.decode("utf-8", errors="replace").rstrip() + if text: + stderr_tail.append(text) + + stderr_task = asyncio.create_task(_drain_stderr()) + + try: + buffer = "" + async for chunk in proc.stdout: + buffer += chunk.decode("utf-8", errors="replace") + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.strip() + if line: + yield line + + if buffer.strip(): + yield buffer.strip() + + await proc.wait() + if proc.returncode: + # The CLI failed (missing binary/auth, bad command). Raise so the + # activity (and turn) surfaces as failed instead of completing with + # no output. Temporal will apply the activity's retry policy. + tail = "\n".join(stderr_tail) + raise RuntimeError( + f"gemini CLI exited with status {proc.returncode}:\n{tail}" + ) + finally: + # Release the subprocess and stderr drain task even if the consumer + # abandons the generator early (task cancellation / client disconnect): + # cancel the drain task and terminate+reap the process if it is still + # running, so neither is leaked. + stderr_task.cancel() + try: + await stderr_task + except asyncio.CancelledError: + pass + if proc.returncode is None: + try: + proc.terminate() + except ProcessLookupError: + pass + await proc.wait() + + +@activity.defn(name=RUN_GEMINI_CLI_TURN_ACTIVITY) +async def run_gemini_cli_turn(params: RunGeminiCliTurnParams) -> dict[str, Any]: + """Run one Gemini CLI turn end-to-end and stream events to the task. + + Runs in an activity (real asyncio loop) so subprocess I/O is permitted. + """ + emitter = UnifiedEmitter( + task_id=params.task_id, + trace_id=params.trace_id, + parent_span_id=params.parent_span_id, + ) + turn = GeminiCliTurn(_spawn_gemini(params.prompt, session_id=params.session_id)) + result = await emitter.auto_send_turn(turn, created_at=params.created_at) + + return RunGeminiCliTurnResult(final_text=result.final_text, session_id=turn.session_id).model_dump() diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/project/run_worker.py.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/run_worker.py.j2 new file mode 100644 index 000000000..6dc6d3323 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/run_worker.py.j2 @@ -0,0 +1,41 @@ +"""Temporal worker for {{ agent_name }} — Gemini CLI harness. + +Run as a separate long-lived process alongside the ACP HTTP server. The +worker polls Temporal for workflow + activity tasks and executes them. + +The Gemini CLI CLI subprocess runs in the ``run_gemini_cli_turn`` activity +(registered below alongside the built-in Agentex activities), because +subprocess I/O is not permitted on the Temporal workflow event loop. +""" + +import asyncio + +from project.workflow import {{ workflow_class }} +from project.activities import run_gemini_cli_turn +from agentex.lib.utils.debug import setup_debug_if_enabled +from agentex.lib.utils.logging import make_logger +from agentex.lib.environment_variables import EnvironmentVariables +from agentex.lib.core.temporal.activities import get_all_activities +from agentex.lib.core.temporal.workers.worker import AgentexWorker + +environment_variables = EnvironmentVariables.refresh() +logger = make_logger(__name__) + + +async def main(): + setup_debug_if_enabled() + + task_queue_name = environment_variables.WORKFLOW_TASK_QUEUE + if task_queue_name is None: + raise ValueError("WORKFLOW_TASK_QUEUE is not set") + + worker = AgentexWorker(task_queue=task_queue_name) + + await worker.run( + activities=[run_gemini_cli_turn, *get_all_activities()], + workflow={{ workflow_class }}, + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/project/workflow.py.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/workflow.py.j2 new file mode 100644 index 000000000..b35c7d6e7 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/project/workflow.py.j2 @@ -0,0 +1,149 @@ +"""Temporal workflow for {{ agent_name }} — Gemini CLI harness. + +Holds per-task state (the last Gemini CLI session_id, for observability) durably across +crashes. Each user message triggers ``on_task_event_send``, which delegates the +turn to the ``run_gemini_cli_turn`` activity. The activity spawns the Gemini +Code CLI, wraps its stdout in ``GeminiCliTurn``, and delivers the turn via +``UnifiedEmitter.auto_send_turn`` (the async Redis push path). + +Note on subprocess inside Temporal +------------------------------------ +Subprocess (and all other) I/O must run in a Temporal *activity*, never in +workflow code. Temporal runs workflow + signal-handler bodies on a +deterministic sandbox event loop that does not implement ``subprocess_exec`` +(spawning the CLI there raises ``NotImplementedError``). The activity also gets +Temporal's retry + timeout guarantees. +""" + +from __future__ import annotations + +import os +import json +import asyncio +from datetime import timedelta + +from temporalio import workflow + +from agentex.lib import adk +from agentex.lib.types.acp import SendEventParams, CreateTaskParams +from agentex.lib.types.tracing import SGPTracingProcessorConfig +from agentex.lib.utils.logging import make_logger +from agentex.types.text_content import TextContent +from agentex.lib.environment_variables import EnvironmentVariables +from agentex.lib.core.temporal.types.workflow import SignalName +from agentex.lib.core.temporal.workflows.workflow import BaseWorkflow +from agentex.lib.core.tracing.tracing_processor_manager import add_tracing_processor_config + +with workflow.unsafe.imports_passed_through(): + from project.activities import RunGeminiCliTurnParams, run_gemini_cli_turn + +add_tracing_processor_config( + SGPTracingProcessorConfig( + sgp_api_key=os.environ.get("SGP_API_KEY", ""), + sgp_account_id=os.environ.get("SGP_ACCOUNT_ID", ""), + sgp_base_url=os.environ.get("SGP_CLIENT_BASE_URL", ""), + ) +) + +environment_variables = EnvironmentVariables.refresh() + +if environment_variables.WORKFLOW_NAME is None: + raise ValueError("Environment variable WORKFLOW_NAME is not set") +if environment_variables.AGENT_NAME is None: + raise ValueError("Environment variable AGENT_NAME is not set") + +logger = make_logger(__name__) + + +@workflow.defn(name=environment_variables.WORKFLOW_NAME) +class {{ workflow_class }}(BaseWorkflow): + """Temporal workflow that runs Gemini CLI locally for each user message. + + Records the Gemini CLI session_id reported by each turn's ``init`` event in + durable workflow state. The Gemini CLI's ``--resume`` takes "latest" or an + index rather than a session id, so each turn runs as an independent prompt; + the id is kept for observability only. + """ + + def __init__(self): + super().__init__(display_name=environment_variables.AGENT_NAME) + self._complete_task = False + self._turn_number = 0 + # Last Gemini CLI session_id (observability only; turns are independent). + self._session_id: str | None = None + # Serialize turns: signal handlers can interleave at await points, so two + # quick messages could both read the same stale _session_id and run + # independent Gemini CLI sessions. The lock keeps turns sequential and + # preserves conversation continuity. + self._turn_lock = asyncio.Lock() + + @workflow.signal(name=SignalName.RECEIVE_EVENT) + async def on_task_event_send(self, params: SendEventParams) -> None: + """Handle a user message: spawn Gemini CLI and push events to the task stream.""" + async with self._turn_lock: + task_id = params.task.id + content = params.event.content + if not isinstance(content, TextContent): + logger.warning("Ignoring non-text event content (type=%s)", getattr(content, "type", "?")) + return + self._turn_number += 1 + prompt = content.content + logger.info("Turn %d for task %s", self._turn_number, task_id) + + await adk.messages.create(task_id=task_id, content=params.event.content) + + async with adk.tracing.span( + trace_id=task_id, + task_id=task_id, + name=f"Turn {self._turn_number}", + input={"message": prompt}, + ) as span: + # Delegate the subprocess turn to an activity: subprocess I/O is not + # permitted on the Temporal workflow event loop. The activity streams + # events to the task and returns the final text + session_id. + # workflow.now() gives a deterministic timestamp under replay. + result = await workflow.execute_activity( + run_gemini_cli_turn, + RunGeminiCliTurnParams( + task_id=task_id, + prompt=prompt, + trace_id=task_id, + parent_span_id=span.id if span else None, + session_id=self._session_id, + created_at=workflow.now(), + ), + # Agentic Gemini CLI runs (multiple tool calls, large codegen) + # can take a while; tune this to your workload. + start_to_close_timeout=timedelta(minutes=30), + ) + + # Record the session_id the CLI reported for this turn. + sid = result.get("session_id") + if sid: + self._session_id = sid + + if span: + span.output = {"final_text": result.get("final_text")} + + @workflow.run + async def on_task_create(self, params: CreateTaskParams) -> str: + logger.info("Task created: %s", params.task.id) + + await adk.messages.create( + task_id=params.task.id, + content=TextContent( + author="agent", + content=( + f"Task initialized with params:\n{json.dumps(params.params, indent=2)}\n" + "Send me a message and I'll run it through Gemini CLI locally." + ), + ), + ) + + await workflow.wait_condition(lambda: self._complete_task, timeout=None) + return "Task completed" + + @workflow.signal + async def complete_task_signal(self) -> None: + logger.info("Received complete_task signal") + self._complete_task = True diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/pyproject.toml.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/pyproject.toml.j2 new file mode 100644 index 000000000..2c6ec9c2f --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/pyproject.toml.j2 @@ -0,0 +1,37 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "{{ project_name }}" +version = "0.1.0" +description = "{{ description }}" +requires-python = ">=3.12" +dependencies = [ + "agentex-sdk", + "scale-gp", + "temporalio>=1.18.2", + "python-dotenv>=1.0,<2", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-asyncio", + "httpx", + "black", + "isort", + "flake8", + "debugpy>=1.8.15", +] + +[tool.hatch.build.targets.wheel] +packages = ["project"] + +[tool.black] +line-length = 88 +target-version = ['py312'] + +[tool.isort] +profile = "black" +line_length = 88 diff --git a/src/agentex/lib/cli/templates/temporal-gemini-cli/requirements.txt.j2 b/src/agentex/lib/cli/templates/temporal-gemini-cli/requirements.txt.j2 new file mode 100644 index 000000000..a060d2331 --- /dev/null +++ b/src/agentex/lib/cli/templates/temporal-gemini-cli/requirements.txt.j2 @@ -0,0 +1,11 @@ +# Agentex SDK +agentex-sdk + +# Scale GenAI Platform Python SDK +scale-gp + +# Temporal workflow engine +temporalio>=1.18.2 + +# Loads .env files for local development +python-dotenv>=1.0,<2 diff --git a/tests/lib/adk/test_gemini_cli_sync.py b/tests/lib/adk/test_gemini_cli_sync.py new file mode 100644 index 000000000..d0f3b81bd --- /dev/null +++ b/tests/lib/adk/test_gemini_cli_sync.py @@ -0,0 +1,192 @@ +"""Tests for the Gemini CLI stream-json -> Agentex StreamTaskMessage* converter.""" + +from __future__ import annotations + +import json +from typing import Any, AsyncIterator + +from agentex.types.text_content import TextContent +from agentex.types.task_message_delta import TextDelta +from agentex.types.task_message_update import ( + StreamTaskMessageDone, + StreamTaskMessageFull, + StreamTaskMessageDelta, + StreamTaskMessageStart, +) +from agentex.types.tool_request_content import ToolRequestContent +from agentex.types.tool_response_content import ToolResponseContent +from agentex.lib.adk._modules._gemini_cli_sync import convert_gemini_cli_to_agentex_events + + +async def _aiter(events: list[Any]) -> AsyncIterator[Any]: + for e in events: + yield e + + +async def _collect(stream: AsyncIterator[Any]) -> list[Any]: + return [e async for e in stream] + + +def _init() -> dict[str, Any]: + return {"type": "init", "timestamp": "t", "session_id": "sess-1", "model": "gemini-2.5-flash"} + + +def _user(text: str) -> dict[str, Any]: + return {"type": "message", "timestamp": "t", "role": "user", "content": text} + + +def _delta(text: str) -> dict[str, Any]: + return {"type": "message", "timestamp": "t", "role": "assistant", "content": text, "delta": True} + + +def _result(**stats: Any) -> dict[str, Any]: + return {"type": "result", "timestamp": "t", "status": "success", "stats": stats} + + +class TestAssistantText: + async def test_delta_chunks_become_one_start_deltas_done(self): + out = await _collect( + convert_gemini_cli_to_agentex_events(_aiter([_init(), _user("hi"), _delta("Hel"), _delta("lo"), _result()])) + ) + assert [type(e) for e in out] == [ + StreamTaskMessageStart, + StreamTaskMessageDelta, + StreamTaskMessageDelta, + StreamTaskMessageDone, + ] + assert isinstance(out[0].content, TextContent) and out[0].content.content == "" + assert isinstance(out[1].delta, TextDelta) and out[1].delta.text_delta == "Hel" + assert out[2].delta.text_delta == "lo" + assert out[0].index == out[1].index == out[2].index == out[3].index + + async def test_user_message_is_ignored(self): + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter([_user("hello"), _result()]))) + assert out == [] + + async def test_non_delta_assistant_message_is_delivered_whole(self): + msg = {"type": "message", "role": "assistant", "content": "Whole answer"} + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter([msg]))) + assert [type(e) for e in out] == [StreamTaskMessageStart, StreamTaskMessageDelta, StreamTaskMessageDone] + assert out[1].delta.text_delta == "Whole answer" + + async def test_materialised_message_closes_open_streamed_slot_without_duplicating(self): + out = await _collect( + convert_gemini_cli_to_agentex_events( + _aiter([_delta("Hel"), _delta("lo"), {"type": "message", "role": "assistant", "content": "Hello"}]) + ) + ) + assert [type(e) for e in out] == [ + StreamTaskMessageStart, + StreamTaskMessageDelta, + StreamTaskMessageDelta, + StreamTaskMessageDone, + ] + + async def test_stream_ending_without_result_still_closes_the_slot(self): + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter([_delta("partial")]))) + assert isinstance(out[-1], StreamTaskMessageDone) + + async def test_raw_json_strings_and_junk_lines(self): + lines = [json.dumps(_delta("A")), "", "not json", json.dumps(_result())] + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter(lines))) + assert [type(e) for e in out] == [StreamTaskMessageStart, StreamTaskMessageDelta, StreamTaskMessageDone] + + +class TestTools: + async def test_tool_use_and_result_pair_by_tool_id(self): + events = [ + _delta("Let me check."), + {"type": "tool_use", "tool_name": "read_file", "tool_id": "call-1", "parameters": {"path": "a.txt"}}, + {"type": "tool_result", "tool_id": "call-1", "status": "success", "output": "file body"}, + _delta("Done."), + _result(), + ] + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter(events))) + kinds = [type(e).__name__ for e in out] + # text slot closed before the tool request; a second slot opened after the result + assert kinds == [ + "StreamTaskMessageStart", + "StreamTaskMessageDelta", + "StreamTaskMessageDone", + "StreamTaskMessageStart", + "StreamTaskMessageDone", + "StreamTaskMessageFull", + "StreamTaskMessageStart", + "StreamTaskMessageDelta", + "StreamTaskMessageDone", + ] + req = out[3].content + assert isinstance(req, ToolRequestContent) + assert req.tool_call_id == "call-1" and req.name == "read_file" and req.arguments == {"path": "a.txt"} + res = out[5].content + assert isinstance(res, ToolResponseContent) + assert res.tool_call_id == "call-1" and res.content == {"result": "file body"} + assert out[3].index == out[4].index and out[5].index not in (out[0].index, out[3].index) + + async def test_error_tool_result_sets_is_error_and_uses_message(self): + events = [ + {"type": "tool_use", "tool_name": "run_shell_command", "tool_id": "call-2", "parameters": {}}, + { + "type": "tool_result", + "tool_id": "call-2", + "status": "error", + "error": {"type": "ToolError", "message": "denied"}, + }, + ] + out = await _collect(convert_gemini_cli_to_agentex_events(_aiter(events))) + full = [e for e in out if isinstance(e, StreamTaskMessageFull)][0] + assert full.content.content == {"result": "denied", "is_error": True} + + async def test_missing_tool_id_gets_a_synthetic_one(self): + out = await _collect( + convert_gemini_cli_to_agentex_events(_aiter([{"type": "tool_use", "tool_name": "x", "parameters": {}}])) + ) + assert out[0].content.tool_call_id == "tool_1" + + +class TestCallbacks: + async def test_on_init_and_on_result_receive_raw_events(self): + seen: dict[str, Any] = {} + + async def on_init(evt: dict[str, Any]) -> None: + seen["init"] = evt + + async def on_result(evt: dict[str, Any]) -> None: + seen["result"] = evt + + await _collect( + convert_gemini_cli_to_agentex_events( + _aiter([_init(), _delta("x"), _result(total_tokens=3)]), on_result=on_result, on_init=on_init + ) + ) + assert seen["init"]["session_id"] == "sess-1" + assert seen["result"]["stats"]["total_tokens"] == 3 + + async def test_error_events_emit_nothing(self): + out = await _collect( + convert_gemini_cli_to_agentex_events( + _aiter([{"type": "error", "severity": "warning", "message": "slow"}, _result()]) + ) + ) + assert out == [] + + async def test_closing_the_generator_closes_the_source(self): + closed = {"v": False} + + class _Src: + def __init__(self) -> None: + self._it = _aiter([_delta("a"), _delta("b"), _result()]) + + def __aiter__(self): + return self + + async def __anext__(self): + return await self._it.__anext__() + + async def aclose(self) -> None: + closed["v"] = True + + gen = convert_gemini_cli_to_agentex_events(_Src()) + await gen.__anext__() + await gen.aclose() + assert closed["v"] is True diff --git a/tests/lib/adk/test_gemini_cli_turn.py b/tests/lib/adk/test_gemini_cli_turn.py new file mode 100644 index 000000000..1c0788deb --- /dev/null +++ b/tests/lib/adk/test_gemini_cli_turn.py @@ -0,0 +1,121 @@ +"""Tests for GeminiCliTurn and gemini_cli_usage_to_turn_usage.""" + +from __future__ import annotations + +from typing import Any, AsyncIterator + +from agentex.lib.core.harness.types import TurnUsage, HarnessTurn +from agentex.types.task_message_update import StreamTaskMessageDone, StreamTaskMessageStart +from agentex.lib.adk._modules._gemini_cli_turn import ( + GeminiCliTurn, + gemini_cli_usage_to_turn_usage, +) + + +async def _aiter(events: list[Any]) -> AsyncIterator[Any]: + for e in events: + yield e + + +def _result(stats: dict[str, Any] | None) -> dict[str, Any]: + evt: dict[str, Any] = {"type": "result", "status": "success"} + if stats is not None: + evt["stats"] = stats + return evt + + +class TestGeminiCliUsageToTurnUsage: + def test_full_stats(self): + usage = gemini_cli_usage_to_turn_usage( + _result( + { + "total_tokens": 30, + "input_tokens": 20, + "output_tokens": 10, + "cached": 5, + "input": 15, + "duration_ms": 1234, + "tool_calls": 2, + "models": {"gemini-2.5-flash": {}}, + } + ) + ) + assert usage.input_tokens == 20 + assert usage.output_tokens == 10 + assert usage.cached_input_tokens == 5 + assert usage.total_tokens == 30 + assert usage.duration_ms == 1234 + assert usage.num_tool_calls == 2 + assert usage.model == "gemini-2.5-flash" + assert usage.cost_usd is None + assert usage.num_llm_calls is None + + def test_explicit_model_wins_over_stats(self): + usage = gemini_cli_usage_to_turn_usage(_result({"models": {"from-stats": {}}}), model="from-init") + assert usage.model == "from-init" + + def test_missing_stats_returns_nones(self): + usage = gemini_cli_usage_to_turn_usage(_result(None)) + assert usage.input_tokens is None and usage.output_tokens is None and usage.total_tokens is None + assert usage.duration_ms is None and usage.num_tool_calls == 0 and usage.model is None + + def test_total_computed_when_absent(self): + usage = gemini_cli_usage_to_turn_usage(_result({"input_tokens": 2, "output_tokens": 3})) + assert usage.total_tokens == 5 + + def test_real_zeros_preserved(self): + usage = gemini_cli_usage_to_turn_usage( + _result({"input_tokens": 0, "output_tokens": 0, "cached": 0, "tool_calls": 0}) + ) + assert usage.input_tokens == 0 and usage.cached_input_tokens == 0 and usage.total_tokens == 0 + + def test_returns_turn_usage_instance(self): + assert isinstance(gemini_cli_usage_to_turn_usage(_result({})), TurnUsage) + + +class TestGeminiCliTurnProtocol: + def test_satisfies_harness_turn_protocol(self): + turn = GeminiCliTurn(_aiter([])) + assert isinstance(turn, HarnessTurn) + + async def test_events_yields_stream_task_messages(self): + turn = GeminiCliTurn( + _aiter([{"type": "message", "role": "assistant", "content": "hi", "delta": True}, _result({})]) + ) + events = [e async for e in turn.events] + assert isinstance(events[0], StreamTaskMessageStart) + assert isinstance(events[-1], StreamTaskMessageDone) + + async def test_usage_before_drain_is_empty(self): + turn = GeminiCliTurn(_aiter([_result({"input_tokens": 1})])) + assert turn.usage() == TurnUsage() + + async def test_usage_after_drain_reflects_result_and_init_model(self): + turn = GeminiCliTurn( + _aiter( + [ + {"type": "init", "session_id": "s-9", "model": "gemini-2.5-pro"}, + _result({"input_tokens": 7, "output_tokens": 1}), + ] + ) + ) + _ = [e async for e in turn.events] + usage = turn.usage() + assert usage.input_tokens == 7 and usage.total_tokens == 8 and usage.model == "gemini-2.5-pro" + assert turn.session_id == "s-9" and turn.model == "gemini-2.5-pro" + + async def test_usage_empty_when_no_result_event(self): + turn = GeminiCliTurn( + _aiter( + [ + {"type": "init", "model": "m"}, + {"type": "message", "role": "assistant", "content": "x", "delta": True}, + ] + ) + ) + _ = [e async for e in turn.events] + assert turn.usage() == TurnUsage(model="m") + + async def test_events_property_returns_same_iterator(self): + turn = GeminiCliTurn(_aiter([])) + assert turn.events is turn.events diff --git a/tests/lib/core/harness/test_harness_gemini_cli_sync.py b/tests/lib/core/harness/test_harness_gemini_cli_sync.py new file mode 100644 index 000000000..a8bd3721a --- /dev/null +++ b/tests/lib/core/harness/test_harness_gemini_cli_sync.py @@ -0,0 +1,98 @@ +"""End-to-end: GeminiCliTurn through UnifiedEmitter.yield_turn (sync HTTP path). + +Checks event order and content, and that tool spans are derived from the +canonical stream the Gemini CLI tap produces. +""" + +from __future__ import annotations + +from typing import Any, AsyncIterator + +from agentex.lib.core.harness.tracer import SpanTracer +from agentex.lib.core.harness.emitter import UnifiedEmitter +from agentex.types.task_message_update import ( + StreamTaskMessageDone, + StreamTaskMessageFull, + StreamTaskMessageStart, +) +from agentex.types.tool_request_content import ToolRequestContent +from agentex.types.tool_response_content import ToolResponseContent +from agentex.lib.adk._modules._gemini_cli_turn import GeminiCliTurn + +from ._fakes import FakeTracing + + +def _tool_then_text_events() -> list[dict[str, Any]]: + return [ + {"type": "init", "session_id": "s", "model": "gemini-2.5-flash"}, + {"type": "message", "role": "user", "content": "What is in a.txt?"}, + {"type": "tool_use", "tool_name": "read_file", "tool_id": "call-1", "parameters": {"path": "a.txt"}}, + {"type": "tool_result", "tool_id": "call-1", "status": "success", "output": "hello"}, + {"type": "message", "role": "assistant", "content": "It says ", "delta": True}, + {"type": "message", "role": "assistant", "content": "hello.", "delta": True}, + {"type": "result", "status": "success", "stats": {"input_tokens": 5, "output_tokens": 3, "tool_calls": 1}}, + ] + + +async def _aiter(events: list[dict[str, Any]]) -> AsyncIterator[dict[str, Any]]: + for e in events: + yield e + + +async def _run_yield_turn( + events: list[dict[str, Any]], + trace_id: str | None = None, + parent_span_id: str | None = None, + fake_tracing: FakeTracing | None = None, +) -> tuple[list[Any], GeminiCliTurn]: + tracer: SpanTracer | bool | None = None + if trace_id and fake_tracing is not None: + tracer = SpanTracer(trace_id=trace_id, parent_span_id=parent_span_id, task_id="task1", tracing=fake_tracing) + turn = GeminiCliTurn(_aiter(events)) + emitter = UnifiedEmitter( + task_id="task1", + trace_id=trace_id, + parent_span_id=parent_span_id, + tracer=tracer if tracer is not None else False, + ) + return [ev async for ev in emitter.yield_turn(turn)], turn + + +class TestSyncYieldEventOrder: + async def test_tool_request_precedes_tool_response_then_text(self) -> None: + out, _ = await _run_yield_turn(_tool_then_text_events()) + kinds = [type(e).__name__ for e in out] + assert kinds.index("StreamTaskMessageFull") > kinds.index("StreamTaskMessageStart") + req = [e for e in out if isinstance(e, StreamTaskMessageStart) and isinstance(e.content, ToolRequestContent)][0] + res = [e for e in out if isinstance(e, StreamTaskMessageFull)][0] + assert isinstance(res.content, ToolResponseContent) + assert req.content.tool_call_id == res.content.tool_call_id == "call-1" + text_start = [ + e for e in out if isinstance(e, StreamTaskMessageStart) and not isinstance(e.content, ToolRequestContent) + ][0] + assert out.index(text_start) > out.index(res) + + async def test_every_start_has_matching_done(self) -> None: + out, _ = await _run_yield_turn(_tool_then_text_events()) + starts = {e.index for e in out if isinstance(e, StreamTaskMessageStart)} + dones = {e.index for e in out if isinstance(e, StreamTaskMessageDone)} + assert starts == dones + + async def test_usage_available_after_turn(self) -> None: + _, turn = await _run_yield_turn(_tool_then_text_events()) + usage = turn.usage() + assert usage.input_tokens == 5 and usage.output_tokens == 3 and usage.num_tool_calls == 1 + assert usage.model == "gemini-2.5-flash" + + +class TestSyncYieldSpanDerivation: + async def test_tool_span_opened_and_closed_with_result(self) -> None: + fake = FakeTracing() + await _run_yield_turn(_tool_then_text_events(), trace_id="trace1", parent_span_id="parent", fake_tracing=fake) + assert "read_file" in fake.started_names + assert any(isinstance(o, dict) and o.get("result") == "hello" for o in fake.ended_outputs) + + async def test_no_trace_id_means_no_spans(self) -> None: + fake = FakeTracing() + await _run_yield_turn(_tool_then_text_events(), trace_id=None, fake_tracing=fake) + assert fake.started == []