torbjorn/my-pal-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat!: Huge update - Link another CLI (such as gemini directly from with Claude Code / Codex). https://github.com/BeehiveInnovations/zen-mcp-server/issues/208

Browse Source 
Zen now allows you to define `roles` for an external CLI and delegate work to another CLI via the new `clink` tool (short for `CLI + Link`). Gemini, for instance, offers 1000 free requests a day - this means you can save on tokens and your weekly limits within Claude Code by delegating work to another entirely capable CLI agent!

Define your own system prompts as `roles` and make another CLI do anything you'd like. Like the current tool you're connected to, the other CLI has complete access to your files and the current context. This also works incredibly well with Zen's `conversation continuity`.
This commit is contained in:
Fahad
2025-10-05 10:40:44 +04:00
parent 0d46976a8a
commit a2ccb48e9a
21 changed files with 1387 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
README.md
View File

@@ -2,6 +2,15 @@
[zen_web.webm](https://github.com/user-attachments/assets/851e3911-7f06-47c0-a4ab-a2601236697c) [zen_web.webm](https://github.com/user-attachments/assets/851e3911-7f06-47c0-a4ab-a2601236697c)
---
### Now with **`clink`** – Connect Your CLIs Together
Need Gemini (or another CLI agent) to join the conversation directly? The new [`clink`](docs/tools/clink.md) tool lets you **connect one CLI to another** ([gemini](https://github.com/google-gemini/gemini-cli) supported for now) so they can **plan, review, debate, and collaborate** inside the same context.
Point `clink` at a CLI like Gemini with role presets (e.g., `planner`, `codereviewer`, `default`) and that agent will handle **web searches, file inspection, and documentation lookups** as a first-class participant in your workflow.
---
<div align="center"> <div align="center">
### Your CLI + Multiple Models = Your AI Dev Team ### Your CLI + Multiple Models = Your AI Dev Team
@@ -157,6 +166,7 @@ cd zen-mcp-server
"Use zen to analyze this code for security issues with gemini pro" "Use zen to analyze this code for security issues with gemini pro"
"Debug this error with o3 and then get flash to suggest optimizations" "Debug this error with o3 and then get flash to suggest optimizations"
"Plan the migration strategy with zen, get consensus from multiple models" "Plan the migration strategy with zen, get consensus from multiple models"
"clink with cli_name=\"gemini\" role=\"planner\" to draft a phased rollout plan"
``` ```
👉 **[Complete Setup Guide](docs/getting-started.md)** with detailed installation, configuration for Gemini / Codex / Qwen, and troubleshooting 👉 **[Complete Setup Guide](docs/getting-started.md)** with detailed installation, configuration for Gemini / Codex / Qwen, and troubleshooting
@@ -171,6 +181,7 @@ Zen activates any provider that has credentials in your `.env`. See `.env.exampl
> **Note:** Each tool comes with its own multi-step workflow, parameters, and descriptions that consume valuable context window space even when not in use. To optimize performance, some tools are disabled by default. See [Tool Configuration](#tool-configuration) below to enable them. > **Note:** Each tool comes with its own multi-step workflow, parameters, and descriptions that consume valuable context window space even when not in use. To optimize performance, some tools are disabled by default. See [Tool Configuration](#tool-configuration) below to enable them.
**Collaboration & Planning** *(Enabled by default)* **Collaboration & Planning** *(Enabled by default)*
- **[`clink`](docs/tools/clink.md)** - Bridge requests to external AI CLIs (Gemini planner, codereviewer, etc.)
- **[`chat`](docs/tools/chat.md)** - Brainstorm ideas, get second opinions, validate approaches - **[`chat`](docs/tools/chat.md)** - Brainstorm ideas, get second opinions, validate approaches
- **[`thinkdeep`](docs/tools/thinkdeep.md)** - Extended reasoning, edge case analysis, alternative perspectives - **[`thinkdeep`](docs/tools/thinkdeep.md)** - Extended reasoning, edge case analysis, alternative perspectives
- **[`planner`](docs/tools/planner.md)** - Break down complex projects into structured, actionable plans - **[`planner`](docs/tools/planner.md)** - Break down complex projects into structured, actionable plans

7
clink/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
"""Public helpers for clink components."""
from __future__ import annotations
from .registry import ClinkRegistry, get_registry
__all__ = ["ClinkRegistry", "get_registry"]

26
clink/agents/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
"""Agent factory for clink CLI integrations."""
from __future__ import annotations
from clink.models import ResolvedCLIClient
from .base import AgentOutput, BaseCLIAgent, CLIAgentError
from .gemini import GeminiAgent
_AGENTS: dict[str, type[BaseCLIAgent]] = {
"gemini": GeminiAgent,
}
def create_agent(client: ResolvedCLIClient) -> BaseCLIAgent:
agent_key = client.name.lower()
agent_cls = _AGENTS.get(agent_key, BaseCLIAgent)
return agent_cls(client)
__all__ = [
"AgentOutput",
"BaseCLIAgent",
"CLIAgentError",
"create_agent",
]

179
clink/agents/base.py Normal file
View File

@@ -0,0 +1,179 @@
"""Execute configured CLI agents for the clink tool and parse output."""
from __future__ import annotations
import asyncio
import logging
import os
import shlex
import tempfile
import time
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
from clink.constants import DEFAULT_STREAM_LIMIT
from clink.models import ResolvedCLIClient, ResolvedCLIRole
from clink.parsers import BaseParser, ParsedCLIResponse, ParserError, get_parser
logger = logging.getLogger("clink.agent")
@dataclass
class AgentOutput:
"""Container returned by CLI agents after successful execution."""
parsed: ParsedCLIResponse
sanitized_command: list[str]
returncode: int
stdout: str
stderr: str
duration_seconds: float
parser_name: str
output_file_content: str | None = None
class CLIAgentError(RuntimeError):
"""Raised when a CLI agent fails (non-zero exit, timeout, parse errors)."""
def __init__(self, message: str, *, returncode: int | None = None, stdout: str = "", stderr: str = "") -> None:
super().__init__(message)
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
class BaseCLIAgent:
"""Execute a configured CLI command and parse its output."""
def __init__(self, client: ResolvedCLIClient):
self.client = client
self._parser: BaseParser = get_parser(client.parser)
self._logger = logging.getLogger(f"clink.runner.{client.name}")
async def run(
self,
*,
role: ResolvedCLIRole,
prompt: str,
files: Sequence[str],
images: Sequence[str],
) -> AgentOutput:
# Files and images are already embedded into the prompt by the tool; they are
# accepted here only to keep parity with SimpleTool callers.
_ = (files, images)
# The runner simply executes the configured CLI command for the selected role.
command = self._build_command(role=role)
env = self._build_environment()
sanitized_command = list(command)
cwd = str(self.client.working_dir) if self.client.working_dir else None
limit = DEFAULT_STREAM_LIMIT
stdout_text = ""
stderr_text = ""
output_file_content: str | None = None
start_time = time.monotonic()
output_file_path: Path | None = None
command_with_output_flag = list(command)
if self.client.output_to_file:
fd, tmp_path = tempfile.mkstemp(prefix="clink-", suffix=".json")
os.close(fd)
output_file_path = Path(tmp_path)
flag_template = self.client.output_to_file.flag_template
try:
rendered_flag = flag_template.format(path=str(output_file_path))
except KeyError as exc: # pragma: no cover - defensive
raise CLIAgentError(f"Invalid output flag template '{flag_template}': missing placeholder {exc}")
command_with_output_flag.extend(shlex.split(rendered_flag))
sanitized_command = list(command_with_output_flag)
self._logger.debug("Executing CLI command: %s", " ".join(sanitized_command))
if cwd:
self._logger.debug("Working directory: %s", cwd)
try:
process = await asyncio.create_subprocess_exec(
*command_with_output_flag,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
limit=limit,
env=env,
)
except FileNotFoundError as exc:
raise CLIAgentError(f"Executable not found for CLI '{self.client.name}': {exc}") from exc
try:
stdout_bytes, stderr_bytes = await asyncio.wait_for(
process.communicate(prompt.encode("utf-8")),
timeout=self.client.timeout_seconds,
)
except asyncio.TimeoutError as exc:
process.kill()
await process.communicate()
raise CLIAgentError(
f"CLI '{self.client.name}' timed out after {self.client.timeout_seconds} seconds",
returncode=None,
) from exc
duration = time.monotonic() - start_time
return_code = process.returncode
stdout_text = stdout_bytes.decode("utf-8", errors="replace")
stderr_text = stderr_bytes.decode("utf-8", errors="replace")
if output_file_path and output_file_path.exists():
output_file_content = output_file_path.read_text(encoding="utf-8", errors="replace")
if self.client.output_to_file and self.client.output_to_file.cleanup:
try:
output_file_path.unlink()
except OSError: # pragma: no cover - best effort cleanup
pass
if output_file_content and not stdout_text.strip():
stdout_text = output_file_content
if return_code != 0:
raise CLIAgentError(
f"CLI '{self.client.name}' exited with status {return_code}",
returncode=return_code,
stdout=stdout_text,
stderr=stderr_text,
)
try:
parsed = self._parser.parse(stdout_text, stderr_text)
except ParserError as exc:
raise CLIAgentError(
f"Failed to parse output from CLI '{self.client.name}': {exc}",
returncode=return_code,
stdout=stdout_text,
stderr=stderr_text,
) from exc
return AgentOutput(
parsed=parsed,
sanitized_command=sanitized_command,
returncode=return_code,
stdout=stdout_text,
stderr=stderr_text,
duration_seconds=duration,
parser_name=self._parser.name,
output_file_content=output_file_content,
)
def _build_command(self, *, role: ResolvedCLIRole) -> list[str]:
base = list(self.client.executable)
base.extend(self.client.internal_args)
base.extend(self.client.config_args)
base.extend(role.role_args)
return base
def _build_environment(self) -> dict[str, str]:
env = os.environ.copy()
env.update(self.client.env)
return env

14
clink/agents/gemini.py Normal file
View File

@@ -0,0 +1,14 @@
"""Gemini-specific CLI agent hooks."""
from __future__ import annotations
from clink.models import ResolvedCLIClient
from .base import BaseCLIAgent
class GeminiAgent(BaseCLIAgent):
"""Placeholder for Gemini-specific behaviour."""
def __init__(self, client: ResolvedCLIClient):
super().__init__(client)

36
clink/constants.py Normal file
View File

@@ -0,0 +1,36 @@
"""Internal defaults and constants for clink."""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
DEFAULT_TIMEOUT_SECONDS = 1800
DEFAULT_STREAM_LIMIT = 10 * 1024 * 1024 # 10MB per stream
PROJECT_ROOT = Path(__file__).resolve().parent.parent
BUILTIN_PROMPTS_DIR = PROJECT_ROOT / "systemprompts" / "clink"
CONFIG_DIR = PROJECT_ROOT / "conf" / "cli_clients"
USER_CONFIG_DIR = Path.home() / ".zen" / "cli_clients"
@dataclass(frozen=True)
class CLIInternalDefaults:
"""Internal defaults applied to a CLI client during registry load."""
parser: str
additional_args: list[str] = field(default_factory=list)
env: dict[str, str] = field(default_factory=dict)
default_role_prompt: str | None = None
timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS
runner: str | None = None
INTERNAL_DEFAULTS: dict[str, CLIInternalDefaults] = {
"gemini": CLIInternalDefaults(
parser="gemini_json",
additional_args=["-o", "json"],
default_role_prompt="systemprompts/clink/gemini_default.txt",
runner="gemini",
),
}

98
clink/models.py Normal file
View File

@@ -0,0 +1,98 @@
"""Pydantic models for clink configuration and runtime structures."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field, PositiveInt, field_validator
class OutputCaptureConfig(BaseModel):
"""Optional configuration for CLIs that write output to disk."""
flag_template: str = Field(..., description="Template used to inject the output path, e.g. '--output {path}'.")
cleanup: bool = Field(
default=True,
description="Whether the temporary file should be removed after reading.",
)
class CLIRoleConfig(BaseModel):
"""Role-specific configuration loaded from JSON manifests."""
prompt_path: str | None = Field(
default=None,
description="Path to the prompt file that seeds this role.",
)
role_args: list[str] = Field(default_factory=list)
description: str | None = Field(default=None)
@field_validator("role_args", mode="before")
@classmethod
def _ensure_list(cls, value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(item) for item in value]
if isinstance(value, str):
return [value]
raise TypeError("role_args must be a list of strings or a single string")
class CLIClientConfig(BaseModel):
"""Raw CLI client configuration before internal defaults are applied."""
name: str
command: str | None = None
working_dir: str | None = None
additional_args: list[str] = Field(default_factory=list)
env: dict[str, str] = Field(default_factory=dict)
timeout_seconds: PositiveInt | None = Field(default=None)
roles: dict[str, CLIRoleConfig] = Field(default_factory=dict)
output_to_file: OutputCaptureConfig | None = None
@field_validator("additional_args", mode="before")
@classmethod
def _ensure_args_list(cls, value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(item) for item in value]
if isinstance(value, str):
return [value]
raise TypeError("additional_args must be a list of strings or a single string")
class ResolvedCLIRole(BaseModel):
"""Runtime representation of a CLI role with resolved prompt path."""
name: str
prompt_path: Path
role_args: list[str] = Field(default_factory=list)
description: str | None = None
class ResolvedCLIClient(BaseModel):
"""Runtime configuration after merging defaults and validating paths."""
name: str
executable: list[str]
working_dir: Path | None
internal_args: list[str] = Field(default_factory=list)
config_args: list[str] = Field(default_factory=list)
env: dict[str, str] = Field(default_factory=dict)
timeout_seconds: int
parser: str
roles: dict[str, ResolvedCLIRole]
output_to_file: OutputCaptureConfig | None = None
def list_roles(self) -> list[str]:
return list(self.roles.keys())
def get_role(self, role_name: str | None) -> ResolvedCLIRole:
key = role_name or "default"
if key not in self.roles:
available = ", ".join(sorted(self.roles.keys()))
raise KeyError(f"Role '{role_name}' not configured for CLI '{self.name}'. Available roles: {available}")
return self.roles[key]

26
clink/parsers/__init__.py Normal file
View File

@@ -0,0 +1,26 @@
"""Parser registry for clink."""
from __future__ import annotations
from .base import BaseParser, ParsedCLIResponse, ParserError
from .gemini import GeminiJSONParser
_PARSER_CLASSES: dict[str, type[BaseParser]] = {
GeminiJSONParser.name: GeminiJSONParser,
}
def get_parser(name: str) -> BaseParser:
normalized = (name or "").lower()
if normalized not in _PARSER_CLASSES:
raise ParserError(f"No parser registered for '{name}'")
parser_cls = _PARSER_CLASSES[normalized]
return parser_cls()
__all__ = [
"BaseParser",
"ParsedCLIResponse",
"ParserError",
"get_parser",
]

27
clink/parsers/base.py Normal file
View File

@@ -0,0 +1,27 @@
"""Parser interfaces for clink runner outputs."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
@dataclass
class ParsedCLIResponse:
"""Result of parsing CLI stdout/stderr."""
content: str
metadata: dict[str, Any]
class ParserError(RuntimeError):
"""Raised when CLI output cannot be parsed into a structured response."""
class BaseParser:
"""Base interface for CLI output parsers."""
name: str = "base"
def parse(self, stdout: str, stderr: str) -> ParsedCLIResponse:
raise NotImplementedError("Parsers must implement parse()")

49
clink/parsers/gemini.py Normal file
View File

@@ -0,0 +1,49 @@
"""Parser for Gemini CLI JSON output."""
from __future__ import annotations
import json
from typing import Any
from .base import BaseParser, ParsedCLIResponse, ParserError
class GeminiJSONParser(BaseParser):
"""Parse stdout produced by `gemini -o json`."""
name = "gemini_json"
def parse(self, stdout: str, stderr: str) -> ParsedCLIResponse:
if not stdout.strip():
raise ParserError("Gemini CLI returned empty stdout while JSON output was expected")
try:
payload: dict[str, Any] = json.loads(stdout)
except json.JSONDecodeError as exc: # pragma: no cover - defensive logging
raise ParserError(f"Failed to decode Gemini CLI JSON output: {exc}") from exc
response = payload.get("response")
if not isinstance(response, str) or not response.strip():
raise ParserError("Gemini CLI response is missing a textual 'response' field")
metadata: dict[str, Any] = {"raw": payload}
stats = payload.get("stats")
if isinstance(stats, dict):
metadata["stats"] = stats
models = stats.get("models")
if isinstance(models, dict) and models:
model_name = next(iter(models.keys()))
metadata["model_used"] = model_name
model_stats = models.get(model_name) or {}
tokens = model_stats.get("tokens")
if isinstance(tokens, dict):
metadata["token_usage"] = tokens
api_stats = model_stats.get("api")
if isinstance(api_stats, dict):
metadata["latency_ms"] = api_stats.get("totalLatencyMs")
if stderr and stderr.strip():
metadata["stderr"] = stderr.strip()
return ParsedCLIResponse(content=response.strip(), metadata=metadata)

252
clink/registry.py Normal file
View File

@@ -0,0 +1,252 @@
"""Configuration registry for clink CLI integrations."""
from __future__ import annotations
import json
import logging
import shlex
from collections.abc import Iterable
from pathlib import Path
from clink.constants import (
CONFIG_DIR,
DEFAULT_TIMEOUT_SECONDS,
INTERNAL_DEFAULTS,
PROJECT_ROOT,
USER_CONFIG_DIR,
CLIInternalDefaults,
)
from clink.models import (
CLIClientConfig,
CLIRoleConfig,
ResolvedCLIClient,
ResolvedCLIRole,
)
from utils.env import get_env
from utils.file_utils import read_json_file
logger = logging.getLogger("clink.registry")
CONFIG_ENV_VAR = "CLI_CLIENTS_CONFIG_PATH"
class RegistryLoadError(RuntimeError):
"""Raised when configuration files are invalid or missing critical data."""
class ClinkRegistry:
"""Loads CLI client definitions and exposes them for schema generation/runtime use."""
def __init__(self) -> None:
self._clients: dict[str, ResolvedCLIClient] = {}
self._load()
def _load(self) -> None:
self._clients.clear()
for config_path in self._iter_config_files():
try:
data = read_json_file(str(config_path))
except json.JSONDecodeError as exc:
raise RegistryLoadError(f"Invalid JSON in {config_path}: {exc}") from exc
if not data:
logger.debug("Skipping empty configuration file: %s", config_path)
continue
config = CLIClientConfig.model_validate(data)
resolved = self._resolve_config(config, source_path=config_path)
key = resolved.name.lower()
if key in self._clients:
logger.info("Overriding CLI configuration for '%s' from %s", resolved.name, config_path)
else:
logger.debug("Loaded CLI configuration for '%s' from %s", resolved.name, config_path)
self._clients[key] = resolved
if not self._clients:
raise RegistryLoadError(
"No CLI clients configured. Ensure conf/cli_clients contains at least one definition or set "
f"{CONFIG_ENV_VAR}."
)
def reload(self) -> None:
"""Reload configurations from disk."""
self._load()
def list_clients(self) -> list[str]:
return sorted(client.name for client in self._clients.values())
def list_roles(self, cli_name: str) -> list[str]:
config = self.get_client(cli_name)
return sorted(config.roles.keys())
def get_client(self, cli_name: str) -> ResolvedCLIClient:
key = cli_name.lower()
if key not in self._clients:
available = ", ".join(self.list_clients())
raise KeyError(f"CLI '{cli_name}' is not configured. Available clients: {available}")
return self._clients[key]
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _iter_config_files(self) -> Iterable[Path]:
search_paths: list[Path] = []
# 1. Built-in configs
search_paths.append(CONFIG_DIR)
# 2. CLI_CLIENTS_CONFIG_PATH environment override (file or directory)
env_path_raw = get_env(CONFIG_ENV_VAR)
if env_path_raw:
env_path = Path(env_path_raw).expanduser()
search_paths.append(env_path)
# 3. User overrides in ~/.zen/cli_clients
search_paths.append(USER_CONFIG_DIR)
seen: set[Path] = set()
for base in search_paths:
if not base:
continue
if base in seen:
continue
seen.add(base)
if base.is_file() and base.suffix.lower() == ".json":
yield base
continue
if base.is_dir():
for path in sorted(base.glob("*.json")):
if path.is_file():
yield path
else:
logger.debug("Configuration path does not exist: %s", base)
def _resolve_config(self, raw: CLIClientConfig, *, source_path: Path) -> ResolvedCLIClient:
if not raw.name:
raise RegistryLoadError(f"CLI configuration at {source_path} is missing a 'name' field")
normalized_name = raw.name.strip()
internal_defaults = INTERNAL_DEFAULTS.get(normalized_name.lower())
if internal_defaults is None:
raise RegistryLoadError(f"CLI '{raw.name}' is not supported by clink")
executable = self._resolve_executable(raw, internal_defaults, source_path)
internal_args = list(internal_defaults.additional_args) if internal_defaults else []
config_args = list(raw.additional_args)
timeout_seconds = raw.timeout_seconds or (
internal_defaults.timeout_seconds if internal_defaults else DEFAULT_TIMEOUT_SECONDS
)
parser_name = internal_defaults.parser
if not parser_name:
raise RegistryLoadError(
f"CLI '{raw.name}' must define a parser either in configuration or internal defaults"
)
env = self._merge_env(raw, internal_defaults)
working_dir = self._resolve_optional_path(raw.working_dir, source_path.parent)
roles = self._resolve_roles(raw, internal_defaults, source_path)
output_to_file = raw.output_to_file
return ResolvedCLIClient(
name=normalized_name,
executable=executable,
internal_args=internal_args,
config_args=config_args,
env=env,
timeout_seconds=int(timeout_seconds),
parser=parser_name,
roles=roles,
output_to_file=output_to_file,
working_dir=working_dir,
)
def _resolve_executable(
self,
raw: CLIClientConfig,
internal_defaults: CLIInternalDefaults | None,
source_path: Path,
) -> list[str]:
command = raw.command
if not command:
raise RegistryLoadError(f"CLI '{raw.name}' must specify a 'command' in configuration")
return shlex.split(command)
def _merge_env(
self,
raw: CLIClientConfig,
internal_defaults: CLIInternalDefaults | None,
) -> dict[str, str]:
merged: dict[str, str] = {}
if internal_defaults and internal_defaults.env:
merged.update(internal_defaults.env)
merged.update(raw.env)
return merged
def _resolve_roles(
self,
raw: CLIClientConfig,
internal_defaults: CLIInternalDefaults | None,
source_path: Path,
) -> dict[str, ResolvedCLIRole]:
roles: dict[str, CLIRoleConfig] = dict(raw.roles)
default_role_prompt = internal_defaults.default_role_prompt if internal_defaults else None
if "default" not in roles:
roles["default"] = CLIRoleConfig(prompt_path=default_role_prompt)
elif roles["default"].prompt_path is None and default_role_prompt:
roles["default"].prompt_path = default_role_prompt
resolved: dict[str, ResolvedCLIRole] = {}
for role_name, role_config in roles.items():
prompt_path_str = role_config.prompt_path or default_role_prompt
if not prompt_path_str:
raise RegistryLoadError(f"Role '{role_name}' for CLI '{raw.name}' must define a prompt_path")
prompt_path = self._resolve_prompt_path(prompt_path_str, source_path.parent)
resolved[role_name] = ResolvedCLIRole(
name=role_name,
prompt_path=prompt_path,
role_args=list(role_config.role_args),
description=role_config.description,
)
return resolved
def _resolve_prompt_path(self, prompt_path: str, base_dir: Path) -> Path:
resolved = self._resolve_path(prompt_path, base_dir)
if not resolved.exists():
raise RegistryLoadError(f"Prompt file not found: {resolved}")
return resolved
def _resolve_optional_path(self, candidate: str | None, base_dir: Path) -> Path | None:
if not candidate:
return None
return self._resolve_path(candidate, base_dir)
def _resolve_path(self, candidate: str, base_dir: Path) -> Path:
path = Path(candidate)
if path.is_absolute():
return path
candidate_path = (base_dir / path).resolve()
if candidate_path.exists():
return candidate_path
project_relative = (PROJECT_ROOT / path).resolve()
return project_relative
_REGISTRY: ClinkRegistry | None = None
def get_registry() -> ClinkRegistry:
global _REGISTRY
if _REGISTRY is None:
_REGISTRY = ClinkRegistry()
return _REGISTRY

20
conf/cli_clients/gemini.json Normal file
View File

@@ -0,0 +1,20 @@
{
"name": "gemini",
"command": "gemini",
"additional_args": ["--telemetry", "false"],
"env": {},
"roles": {
"default": {
"prompt_path": "systemprompts/clink/gemini_default.txt",
"role_args": []
},
"planner": {
"prompt_path": "systemprompts/clink/gemini_planner.txt",
"role_args": ["--experimental-acp"]
},
"codereviewer": {
"prompt_path": "systemprompts/clink/gemini_codereviewer.txt",
"role_args": []
}
}
}

150
docs/tools/clink.md Normal file
View File

@@ -0,0 +1,150 @@
# Clink Tool - CLI-to-CLI Bridge
**Bring other AI CLIs into your workflow - Gemini (for now), Qwen (soon), Codex (soon), and more work alongside Claude without context switching**
The `clink` tool lets you leverage external AI CLIs (like Gemini CLI, etc.) directly within your current conversation. Instead of switching between terminal windows or losing context, you can ask Gemini to plan a complex migration, review code with specialized prompts, or answer questions - all while staying in your primary Claude Code workflow.
## Why Use Clink (CLI + Link)?
**Scenario 1**: You're working in Claude Code and want Gemini's 1M context window to analyze a massive codebase, or you need Gemini's latest web search to validate documentation.
**Without clink**: Open a new terminal, run `gemini`, lose your conversation context, manually copy/paste findings back.
**With clink**: Just say `"clink with gemini to review this entire codebase for architectural issues"` - Gemini launches separately, processes request and returns results, and the conversation continues seamlessly.
**Scenario 2**: Use [`consensus`](consensus.md) to debate which feature to implement next with multiple models, then seamlessly hand off to Gemini for implementation.
```
"Use consensus with pro and gpt5 to decide whether to add dark mode or offline support next"
[consensus runs, models deliberate, recommendation emerges]
"Continue with clink - implement the recommended feature"
```
Gemini receives the full conversation context from `consensus` including the consensus prompt + replies, understands the chosen feature, technical constraints discussed, and can start implementation immediately. No re-explaining, no context loss - true conversation continuity across tools and models.
## Key Features
- **Stay in one CLI**: No switching between terminal sessions or losing context
- **Full conversation continuity**: Gemini's responses participate in the same conversation thread
- **Role-based prompts**: Pre-configured roles for planning, code review, or general questions
- **Full CLI capabilities**: Gemini can use its own web search, file tools, and latest features
- **Token efficiency**: File references (not full content) to conserve tokens
- **Cross-tool collaboration**: Combine with other Zen tools like `planner` → `clink` → `codereview`
- **Free tier available**: Gemini offers 1,000 requests/day free with a personal Google account - great for cost savings across tools
## Available Roles
**Default Role** - General questions, summaries, quick answers
```
"Use clink to ask gemini about the latest React 19 features"
```
**Planner Role** - Strategic planning with multi-phase approach
```
"Clink with gemini role='planner' to map out our microservices migration strategy"
```
**Code Reviewer Role** - Focused code analysis with severity levels
```
"Use clink role='codereviewer' to review auth.py for security issues"
```
You can make your own custom roles in `conf/cli_clients/gemini.json` or tweak existing ones.
## Tool Parameters
- `prompt`: Your question or task for the external CLI (required)
- `cli_name`: Which CLI to use - `gemini` (default), or add your own in `conf/cli_clients/`
- `role`: Preset role - `default`, `planner`, `codereviewer` (default: `default`)
- `files`: Optional file paths for context (references only, CLI opens files itself)
- `images`: Optional image paths for visual context
- `continuation_id`: Continue previous clink conversations
## Usage Examples
**Architecture Planning:**
```
"Use clink with gemini planner to design a 3-phase rollout plan for our feature flags system"
```
**Code Review with Context:**
```
"Clink to gemini codereviewer: Review payment_service.py for race conditions and concurrency issues"
```
**Quick Research Question:**
```
"Ask gemini via clink: What are the breaking changes in TypeScript 5.5?"
```
**Multi-Tool Workflow:**
```
"Use planner to outline the refactor, then clink gemini planner for validation,
then codereview to verify the implementation"
```
**Leveraging Gemini's Web Search:**
```
"Clink gemini to research current best practices for Kubernetes autoscaling in 2025"
```
## How Clink Works
1. **Your request** - You ask your current CLI to use `clink` with a specific CLI and role
2. **Background execution** - Zen spawns the configured CLI (e.g., `gemini --output-format json`)
3. **Context forwarding** - Your prompt, files (as references), and conversation history are sent as part of the prompt
4. **CLI processing** - Gemini (or other CLI) uses its own tools: web search, file access, thinking modes
5. **Seamless return** - Results flow back into your conversation with full context preserved
6. **Continuation support** - Future tools and models can reference Gemini's findings via [continuation support](../context-revival.md) within Zen.
## Best Practices
- **Pre-authenticate CLIs**: Install and configure Gemini CLI first (`npm install -g @google/gemini-cli`)
- **Choose appropriate roles**: Use `planner` for strategy, `codereviewer` for code, `default` for general questions
- **Leverage CLI strengths**: Gemini's 1M context for large codebases, web search for current docs
- **Combine with Zen tools**: Chain `clink` with `planner`, `codereview`, `debug` for powerful workflows
- **File efficiency**: Pass file paths, let the CLI decide what to read (saves tokens)
## Configuration
Clink configurations live in `conf/cli_clients/`. The default `gemini.json` includes:
```json
{
"name": "gemini",
"command": "gemini",
"additional_args": ["--telemetry", "false"],
"roles": {
"planner": {
"prompt_path": "systemprompts/clink/gemini_planner.txt",
"role_args": ["--experimental-acp"]
}
}
}
```
**Adding new CLIs**: Drop a JSON config into `conf/cli_clients/` and create role prompts in `systemprompts/clink/`.
## When to Use Clink vs Other Tools
- **Use `clink`** for: Leveraging external CLI capabilities (Gemini's web search, 1M context), specialized CLI features, cross-CLI collaboration
- **Use `chat`** for: Direct model-to-model conversations within Zen
- **Use `planner`** for: Zen's native planning workflows with step validation
- **Use `codereview`** for: Zen's structured code review with severity levels
**CAUTION**: `clink` opens additional doors but not without additional risk. Configuration arguments like `--approval-mode yolo` in `conf/cli_clients/gemini.json` bypass safety
prompts and should only be used when you're certain the operations are safe.
Review your role configurations carefully before enabling automated execution modes. You can add safeguards and guardrails within role-specific system prompts (in `systemprompts/clink/`) to restrict or guide the external CLI's behavior.
## Setup Requirements
Ensure [gemini](https://github.com/google-gemini/gemini-cli) is installed and configured.
## Related Guides
- [Chat Tool](chat.md) - Direct model conversations
- [Planner Tool](planner.md) - Zen's native planning workflows
- [CodeReview Tool](codereview.md) - Structured code reviews
- [Context Revival](../context-revival.md) - Continuing conversations across tools
- [Advanced Usage](../advanced-usage.md) - Complex multi-tool workflows

7
server.py
View File

@@ -51,6 +51,7 @@ from tools import ( # noqa: E402
AnalyzeTool, AnalyzeTool,
ChallengeTool, ChallengeTool,
ChatTool, ChatTool,
CLinkTool,
CodeReviewTool, CodeReviewTool,
ConsensusTool, ConsensusTool,
DebugIssueTool, DebugIssueTool,
@@ -257,6 +258,7 @@ def filter_disabled_tools(all_tools: dict[str, Any]) -> dict[str, Any]:
# Tools are instantiated once and reused across requests (stateless design) # Tools are instantiated once and reused across requests (stateless design)
TOOLS = { TOOLS = {
"chat": ChatTool(), # Interactive development chat and brainstorming "chat": ChatTool(), # Interactive development chat and brainstorming
"clink": CLinkTool(), # Bridge requests to configured AI CLIs
"thinkdeep": ThinkDeepTool(), # Step-by-step deep thinking workflow with expert analysis "thinkdeep": ThinkDeepTool(), # Step-by-step deep thinking workflow with expert analysis
"planner": PlannerTool(), # Interactive sequential planner using workflow architecture "planner": PlannerTool(), # Interactive sequential planner using workflow architecture
"consensus": ConsensusTool(), # Step-by-step consensus workflow with multi-model analysis "consensus": ConsensusTool(), # Step-by-step consensus workflow with multi-model analysis
@@ -282,6 +284,11 @@ PROMPT_TEMPLATES = {
"description": "Chat and brainstorm ideas", "description": "Chat and brainstorm ideas",
"template": "Chat with {model} about this", "template": "Chat with {model} about this",
}, },
"clink": {
"name": "clink",
"description": "Forward a request to a configured AI CLI (e.g., Gemini)",
"template": "Use clink with cli_name=<cli> to run this prompt",
},
"thinkdeep": { "thinkdeep": {
"name": "thinkdeeper", "name": "thinkdeeper",
"description": "Step-by-step deep thinking workflow with expert analysis", "description": "Step-by-step deep thinking workflow with expert analysis",

7
systemprompts/clink/gemini_codereviewer.txt Normal file
View File

@@ -0,0 +1,7 @@
You are the Gemini CLI code reviewer for the Clink tool.
- Inspect any relevant files directly—use your full repository access, gather whatever context you require before writing feedback.
- Report findings in severity order (Critical, High, Medium, Low) across security, correctness, performance, maintainability; stay anchored to the current change scope.
- For each issue cite precise references (full-file-path:line plus a short excerpt or symbol name), explain the impact, and propose a concrete fix or mitigation.
- Call out positive practices worth retaining so peers know what to preserve.
- Keep feedback precise, actionable, and tailored—avoid speculative refactors or unrelated suggestions.

6
systemprompts/clink/gemini_default.txt Normal file
View File

@@ -0,0 +1,6 @@
You are the Gemini CLI agent operating inside the Zen MCP server with full repository access.
- Use your tools to inspect files, and gather context before responding; quote exact paths, symbols, or commands when they matter.
- Produce clear, direct answers in Markdown tailored to engineers working from the CLI, and highlight actionable next steps.
- Call out assumptions, missing inputs, or follow-up work that would improve confidence in the result.
- If a request is unsafe, infeasible, or violates policy, explain why and provide a safer alternative or mitigation.

7
systemprompts/clink/gemini_planner.txt Normal file
View File

@@ -0,0 +1,7 @@
You are the Gemini CLI planner for the Clink tool.
- Use your full repository access to inspect any relevant files, scripts, or docs before detailing the plan.
- Break objectives into numbered phases with dependencies, validation gates, alternatives, and clear next actions; highlight risks with mitigations.
- Cite concrete references—file paths, line numbers, function or class names—whenever you point to source context.
- Branch when multiple viable strategies exist and explain when to choose each.
- When planning completes, present a polished summary with ASCII visuals, checklists, and guidance another engineer can execute.

42
tests/test_clink_integration.py Normal file
View File

@@ -0,0 +1,42 @@
import json
import os
import shutil
import pytest
from tools.clink import CLinkTool
@pytest.mark.integration
@pytest.mark.asyncio
async def test_clink_gemini_single_digit_sum():
if shutil.which("gemini") is None:
pytest.skip("gemini CLI is not installed or on PATH")
if not (os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")):
pytest.skip("Gemini API key is not configured")
tool = CLinkTool()
prompt = "Respond with a single digit equal to the sum of 2 + 2. Output only that digit."
results = await tool.execute(
{
"prompt": prompt,
"cli_name": "gemini",
"role": "default",
"files": [],
"images": [],
}
)
assert results, "clink tool returned no outputs"
payload = json.loads(results[0].text)
status = payload["status"]
assert status in {"success", "continuation_available"}
content = payload.get("content", "").strip()
assert content == "4"
if status == "continuation_available":
offer = payload.get("continuation_offer") or {}
assert offer.get("continuation_id"), "Expected continuation metadata when status indicates availability"

94
tests/test_clink_tool.py Normal file
View File

@@ -0,0 +1,94 @@
import json
import pytest
from clink import get_registry
from clink.agents import AgentOutput
from clink.parsers.base import ParsedCLIResponse
from tools.clink import CLinkTool
@pytest.mark.asyncio
async def test_clink_tool_execute(monkeypatch):
tool = CLinkTool()
async def fake_run(**kwargs):
return AgentOutput(
parsed=ParsedCLIResponse(content="Hello from Gemini", metadata={"model_used": "gemini-2.5-pro"}),
sanitized_command=["gemini", "-o", "json"],
returncode=0,
stdout='{"response": "Hello from Gemini"}',
stderr="",
duration_seconds=0.42,
parser_name="gemini_json",
output_file_content=None,
)
class DummyAgent:
async def run(self, **kwargs):
return await fake_run(**kwargs)
def fake_create_agent(client):
return DummyAgent()
monkeypatch.setattr("tools.clink.create_agent", fake_create_agent)
arguments = {
"prompt": "Summarize the project",
"cli_name": "gemini",
"role": "default",
"files": [],
"images": [],
}
results = await tool.execute(arguments)
assert len(results) == 1
payload = json.loads(results[0].text)
assert payload["status"] in {"success", "continuation_available"}
assert "Hello from Gemini" in payload["content"]
metadata = payload.get("metadata", {})
assert metadata.get("cli_name") == "gemini"
assert metadata.get("command") == ["gemini", "-o", "json"]
def test_registry_lists_roles():
registry = get_registry()
clients = registry.list_clients()
assert "gemini" in clients
roles = registry.list_roles("gemini")
assert "default" in roles
@pytest.mark.asyncio
async def test_clink_tool_defaults_to_first_cli(monkeypatch):
tool = CLinkTool()
async def fake_run(**kwargs):
return AgentOutput(
parsed=ParsedCLIResponse(content="Default CLI response", metadata={}),
sanitized_command=["gemini"],
returncode=0,
stdout='{"response": "Default CLI response"}',
stderr="",
duration_seconds=0.1,
parser_name="gemini_json",
output_file_content=None,
)
class DummyAgent:
async def run(self, **kwargs):
return await fake_run(**kwargs)
monkeypatch.setattr("tools.clink.create_agent", lambda client: DummyAgent())
arguments = {
"prompt": "Hello",
"files": [],
"images": [],
}
result = await tool.execute(arguments)
payload = json.loads(result[0].text)
metadata = payload.get("metadata", {})
assert metadata.get("cli_name") == tool._default_cli_name

2
tools/__init__.py
View File

@@ -5,6 +5,7 @@ Tool implementations for Zen MCP Server
from .analyze import AnalyzeTool from .analyze import AnalyzeTool
from .challenge import ChallengeTool from .challenge import ChallengeTool
from .chat import ChatTool from .chat import ChatTool
from .clink import CLinkTool
from .codereview import CodeReviewTool from .codereview import CodeReviewTool
from .consensus import ConsensusTool from .consensus import ConsensusTool
from .debug import DebugIssueTool from .debug import DebugIssueTool
@@ -26,6 +27,7 @@ __all__ = [
"DocgenTool", "DocgenTool",
"AnalyzeTool", "AnalyzeTool",
"ChatTool", "ChatTool",
"CLinkTool",
"ConsensusTool", "ConsensusTool",
"ListModelsTool", "ListModelsTool",
"PlannerTool", "PlannerTool",

327
tools/clink.py Normal file
View File

@@ -0,0 +1,327 @@
"""clink tool - bridge Zen MCP requests to external AI CLIs."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from mcp.types import TextContent
from pydantic import BaseModel, Field
from clink import get_registry
from clink.agents import AgentOutput, CLIAgentError, create_agent
from clink.models import ResolvedCLIClient, ResolvedCLIRole
from config import TEMPERATURE_BALANCED
from tools.models import ToolModelCategory, ToolOutput
from tools.shared.base_models import COMMON_FIELD_DESCRIPTIONS
from tools.simple.base import SchemaBuilder, SimpleTool
logger = logging.getLogger(__name__)
class CLinkRequest(BaseModel):
"""Request model for clink tool."""
prompt: str = Field(..., description="Prompt forwarded to the target CLI.")
cli_name: str | None = Field(
default=None,
description="Configured CLI client name to invoke. Defaults to the first configured CLI if omitted.",
)
role: str | None = Field(
default=None,
description="Optional role preset defined in the CLI configuration (defaults to 'default').",
)
files: list[str] = Field(
default_factory=list,
description=COMMON_FIELD_DESCRIPTIONS["files"],
)
images: list[str] = Field(
default_factory=list,
description=COMMON_FIELD_DESCRIPTIONS["images"],
)
continuation_id: str | None = Field(
default=None,
description=COMMON_FIELD_DESCRIPTIONS["continuation_id"],
)
class CLinkTool(SimpleTool):
"""Bridge MCP requests to configured CLI agents.
Schema metadata is cached at construction time and execution relies on the shared
SimpleTool hooks for conversation memory. Prompt preparation is customised so we
pass instructions and file references suitable for another CLI agent.
"""
def __init__(self) -> None:
# Cache registry metadata so the schema surfaces concrete enum values.
self._registry = get_registry()
self._cli_names = self._registry.list_clients()
self._role_map: dict[str, list[str]] = {name: self._registry.list_roles(name) for name in self._cli_names}
self._all_roles: list[str] = sorted({role for roles in self._role_map.values() for role in roles})
self._default_cli_name: str | None = self._cli_names[0] if self._cli_names else None
self._active_system_prompt: str = ""
super().__init__()
def get_name(self) -> str:
return "clink"
def get_description(self) -> str:
return (
"Link a request to an external AI CLI (Gemini CLI, Qwen CLI, etc.) through Zen MCP to reuse "
"their capabilities inside existing workflows."
)
def get_annotations(self) -> dict[str, Any]:
return {"readOnlyHint": True}
def requires_model(self) -> bool:
return False
def get_model_category(self) -> ToolModelCategory:
return ToolModelCategory.BALANCED
def get_default_temperature(self) -> float:
return TEMPERATURE_BALANCED
def get_system_prompt(self) -> str:
return self._active_system_prompt or ""
def get_request_model(self):
return CLinkRequest
def get_input_schema(self) -> dict[str, Any]:
# Surface configured CLI names and roles directly in the schema so MCP clients
# (and downstream agents) can discover available options without consulting
# a separate registry call.
role_descriptions = []
for name in self._cli_names:
roles = ", ".join(sorted(self._role_map.get(name, ["default"]))) or "default"
role_descriptions.append(f"{name}: {roles}")
if role_descriptions:
cli_available = ", ".join(self._cli_names) if self._cli_names else "(none configured)"
default_text = (
f" Default: {self._default_cli_name}." if self._default_cli_name and len(self._cli_names) <= 1 else ""
)
cli_description = (
"Configured CLI client name (from conf/cli_clients). Available: " + cli_available + default_text
)
role_description = (
"Optional role preset defined for the selected CLI (defaults to 'default'). Roles per CLI: "
+ "; ".join(role_descriptions)
)
else:
cli_description = "Configured CLI client name (from conf/cli_clients)."
role_description = "Optional role preset defined for the selected CLI (defaults to 'default')."
properties = {
"prompt": {
"type": "string",
"description": "User request forwarded to the CLI (conversation context is pre-applied).",
},
"cli_name": {
"type": "string",
"enum": self._cli_names,
"description": cli_description,
},
"role": {
"type": "string",
"enum": self._all_roles or ["default"],
"description": role_description,
},
"files": SchemaBuilder.SIMPLE_FIELD_SCHEMAS["files"],
"images": SchemaBuilder.COMMON_FIELD_SCHEMAS["images"],
"continuation_id": SchemaBuilder.COMMON_FIELD_SCHEMAS["continuation_id"],
}
schema = {
"type": "object",
"properties": properties,
"required": ["prompt"],
"additionalProperties": False,
}
if len(self._cli_names) > 1:
schema["required"].append("cli_name")
return schema
def get_tool_fields(self) -> dict[str, dict[str, Any]]:
"""Unused by clink because we override the schema end-to-end."""
return {}
async def execute(self, arguments: dict[str, Any]) -> list[TextContent]:
self._current_arguments = arguments
request = self.get_request_model()(**arguments)
path_error = self._validate_file_paths(request)
if path_error:
return [self._error_response(path_error)]
selected_cli = request.cli_name or self._default_cli_name
if not selected_cli:
return [self._error_response("No CLI clients are configured for clink.")]
try:
client_config = self._registry.get_client(selected_cli)
except KeyError as exc:
return [self._error_response(str(exc))]
try:
role_config = client_config.get_role(request.role)
except KeyError as exc:
return [self._error_response(str(exc))]
files = self.get_request_files(request)
images = self.get_request_images(request)
continuation_id = self.get_request_continuation_id(request)
self._model_context = arguments.get("_model_context")
try:
prompt_text = await self._prepare_prompt_for_role(request, role_config)
except Exception as exc:
logger.exception("Failed to prepare clink prompt")
return [self._error_response(f"Failed to prepare prompt: {exc}")]
agent = create_agent(client_config)
try:
result = await agent.run(role=role_config, prompt=prompt_text, files=files, images=images)
except CLIAgentError as exc:
metadata = self._build_error_metadata(client_config, exc)
error_output = ToolOutput(
status="error",
content=f"CLI '{client_config.name}' execution failed: {exc}",
content_type="text",
metadata=metadata,
)
return [TextContent(type="text", text=error_output.model_dump_json())]
model_info = {
"provider": client_config.name,
"model_name": result.parsed.metadata.get("model_used"),
}
if continuation_id:
try:
self._record_assistant_turn(continuation_id, result.parsed.content, request, model_info)
except Exception:
logger.debug("Failed to record assistant turn for continuation %s", continuation_id, exc_info=True)
metadata = self._build_success_metadata(client_config, role_config, result)
continuation_offer = self._create_continuation_offer(request, model_info)
if continuation_offer:
tool_output = self._create_continuation_offer_response(
result.parsed.content,
continuation_offer,
request,
model_info,
)
tool_output.metadata = self._merge_metadata(tool_output.metadata, metadata)
else:
tool_output = ToolOutput(
status="success",
content=result.parsed.content,
content_type="text",
metadata=metadata,
)
return [TextContent(type="text", text=tool_output.model_dump_json())]
async def prepare_prompt(self, request) -> str:
client_config = self._registry.get_client(request.cli_name)
role_config = client_config.get_role(request.role)
return await self._prepare_prompt_for_role(request, role_config)
async def _prepare_prompt_for_role(self, request: CLinkRequest, role: ResolvedCLIRole) -> str:
"""Load the role prompt and assemble the final user message."""
self._active_system_prompt = role.prompt_path.read_text(encoding="utf-8")
try:
user_content = self.handle_prompt_file_with_fallback(request).strip()
guidance = self._agent_capabilities_guidance()
file_section = self._format_file_references(self.get_request_files(request))
sections: list[str] = []
active_prompt = self.get_system_prompt().strip()
if active_prompt:
sections.append(active_prompt)
sections.append(guidance)
sections.append("=== USER REQUEST ===\n" + user_content)
if file_section:
sections.append("=== FILE REFERENCES ===\n" + file_section)
sections.append("Provide your response below using your own CLI tools as needed:")
return "\n\n".join(sections)
finally:
self._active_system_prompt = ""
def _merge_metadata(self, base: dict[str, Any] | None, extra: dict[str, Any]) -> dict[str, Any]:
merged = dict(base or {})
merged.update(extra)
return merged
def _build_success_metadata(
self,
client: ResolvedCLIClient,
role: ResolvedCLIRole,
result: AgentOutput,
) -> dict[str, Any]:
"""Capture execution metadata for successful CLI calls."""
metadata: dict[str, Any] = {
"cli_name": client.name,
"role": role.name,
"command": result.sanitized_command,
"duration_seconds": round(result.duration_seconds, 3),
"parser": result.parser_name,
"return_code": result.returncode,
}
metadata.update(result.parsed.metadata)
if result.stderr.strip():
metadata.setdefault("stderr", result.stderr.strip())
if result.output_file_content and "raw" not in metadata:
metadata["raw_output_file"] = result.output_file_content
return metadata
def _build_error_metadata(self, client: ResolvedCLIClient, exc: CLIAgentError) -> dict[str, Any]:
"""Assemble metadata for failed CLI calls."""
metadata: dict[str, Any] = {
"cli_name": client.name,
"return_code": exc.returncode,
}
if exc.stdout:
metadata["stdout"] = exc.stdout.strip()
if exc.stderr:
metadata["stderr"] = exc.stderr.strip()
return metadata
def _error_response(self, message: str) -> TextContent:
error_output = ToolOutput(status="error", content=message, content_type="text")
return TextContent(type="text", text=error_output.model_dump_json())
def _agent_capabilities_guidance(self) -> str:
return (
"You are operating through the Gemini CLI agent. You have access to your full suite of "
"CLI capabilities—including launching web searches, reading files, and using any other "
"available tools. Gather current information yourself and deliver the final answer without "
"asking the Zen MCP host to perform searches or file reads."
)
def _format_file_references(self, files: list[str]) -> str:
if not files:
return ""
references: list[str] = []
for file_path in files:
try:
path = Path(file_path)
stat = path.stat()
modified = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
size = stat.st_size
references.append(f"- {file_path} (last modified {modified}, {size} bytes)")
except OSError:
references.append(f"- {file_path} (unavailable)")
return "\n".join(references)