CLI Agent MCP

Instructions

Implementation Reference

• src/cli_agent_mcp/server.py:207-209 (registration)

  Registration of the 'banana' tool: In the MCP server's call_tool method, instantiates BananaHandler when base_name == 'banana'.
  if base_name == "banana": handler = BananaHandler() return await handler.handle(arguments, tool_ctx)
• src/cli_agent_mcp/handlers/image_tools.py:23-173 (handler)

  BananaHandler: MCP tool handler for 'banana'. Defines name, description, input schema, validation, and orchestrates execution via BananaInvoker, GUI events, and response formatting.
  class BananaHandler(ToolHandler): """Banana 图像生成工具处理器。""" @property def name(self) -> str: return "banana" @property def description(self) -> str: return """Generate images using Nano Banana Pro (Gemini 3 Pro Image). CAPABILITIES: - Text-to-image generation with high quality output - Image editing and transformation with reference images - Multiple aspect ratios and resolutions (1K/2K/4K) - Style transfer and multi-image fusion - Optional search grounding for factual content RESPONSE FORMAT: - Returns XML with file paths to generated images - Images are saved to disk (no base64 in response) - Includes text descriptions and optional thinking process BEST PRACTICES: - Be descriptive: describe scenes, not just keywords - Use negative constraints in prompt: "no text", "no watermark" - For editing: provide reference image and specify what to keep - For style transfer: provide style reference image Supports: reference images with roles (edit_base, style_ref, etc.).""" def get_input_schema(self) -> dict[str, Any]: return { "type": "object", "properties": { "prompt": {"type": "string", "description": "Image generation prompt"}, "save_path": {"type": "string", "description": "Base directory for saving images"}, "task_note": {"type": "string", "description": "Subdirectory name for saving images"}, "images": {"type": "array", "items": {"type": "object"}, "default": []}, "aspect_ratio": {"type": "string", "default": "1:1"}, "resolution": {"type": "string", "default": "4K", "enum": ["1K", "2K", "4K"]}, "use_search": {"type": "boolean", "default": False}, "include_thoughts": {"type": "boolean", "default": False}, "temperature": {"type": "number", "default": 1.0}, "top_p": {"type": "number", "default": 0.95}, "top_k": {"type": "integer", "default": 40}, "num_images": {"type": "integer", "default": 1, "minimum": 1, "maximum": 4}, "debug": {"type": "boolean", "description": "Enable debug output"}, }, "required": ["prompt", "save_path", "task_note"], } def validate(self, arguments: dict[str, Any]) -> str | None: if not arguments.get("prompt"): return "Missing required argument: 'prompt'" if not arguments.get("save_path"): return "Missing required argument: 'save_path'" if not arguments.get("task_note"): return "Missing required argument: 'task_note'" return None async def handle( self, arguments: dict[str, Any], ctx: ToolContext, ) -> list[TextContent]: prompt = arguments.get("prompt", "") task_note = arguments.get("task_note", "") # 推送用户 prompt 到 GUI ctx.push_user_prompt("banana", prompt, task_note) # 创建事件回调 def event_callback(event: Any) -> None: if ctx.gui_manager and ctx.gui_manager.is_running: event_dict = event.model_dump() if hasattr(event, "model_dump") else dict(event.__dict__) event_dict["source"] = "banana" ctx.gui_manager.push_event(event_dict) # 创建 invoker 并执行 invoker = BananaInvoker(event_callback=event_callback) params = BananaParams( prompt=prompt, images=arguments.get("images", []), aspect_ratio=arguments.get("aspect_ratio", "1:1"), resolution=arguments.get("resolution", "4K"), use_search=arguments.get("use_search", False), include_thoughts=arguments.get("include_thoughts", False), temperature=arguments.get("temperature", 1.0), top_p=arguments.get("top_p", 0.95), top_k=arguments.get("top_k", 40), num_images=arguments.get("num_images", 1), save_path=arguments.get("save_path", ""), task_note=task_note, ) try: result = await invoker.execute(params) if result.success: response = result.response_xml # 添加 debug_info（仅当 debug 开启时） debug_enabled = ctx.resolve_debug(arguments) if debug_enabled and result.artifacts: response += ( f"\n<debug_info>" f"\n <image_count>{len(result.artifacts)}</image_count>" f"\n <duration_sec>{result.duration_sec:.3f}</duration_sec>" f"\n <model>{result.model}</model>" f"\n <api_endpoint>{result.api_endpoint}</api_endpoint>" f"\n <auth_token>{result.auth_token_masked}</auth_token>" f"\n</debug_info>" ) # 推送结果到 GUI gui_metadata: dict[str, Any] = { "artifacts": result.artifacts, "task_note": task_note, } if debug_enabled: gui_metadata["debug"] = { "image_count": len(result.artifacts), "duration_sec": result.duration_sec, "model": result.model, "api_endpoint": result.api_endpoint, "auth_token": result.auth_token_masked, } ctx.push_to_gui({ "category": "operation", "operation_type": "tool_call", "source": "banana", "session_id": f"banana_{result.request_id}", "name": "banana", "status": "success", "output": response, "metadata": gui_metadata, }) return [TextContent(type="text", text=response)] else: return format_error_response(result.error or "Unknown error") except asyncio.CancelledError: raise except Exception as e: logger.exception(f"Banana tool error: {e}") return format_error_response(str(e))
• src/cli_agent_mcp/handlers/image_tools.py:54-72 (schema)

  Input schema definition for the 'banana' tool, specifying parameters like prompt, save_path, images, aspect_ratio, etc.
  def get_input_schema(self) -> dict[str, Any]: return { "type": "object", "properties": { "prompt": {"type": "string", "description": "Image generation prompt"}, "save_path": {"type": "string", "description": "Base directory for saving images"}, "task_note": {"type": "string", "description": "Subdirectory name for saving images"}, "images": {"type": "array", "items": {"type": "object"}, "default": []}, "aspect_ratio": {"type": "string", "default": "1:1"}, "resolution": {"type": "string", "default": "4K", "enum": ["1K", "2K", "4K"]}, "use_search": {"type": "boolean", "default": False}, "include_thoughts": {"type": "boolean", "default": False}, "temperature": {"type": "number", "default": 1.0}, "top_p": {"type": "number", "default": 0.95}, "top_k": {"type": "integer", "default": 40}, "num_images": {"type": "integer", "default": 1, "minimum": 1, "maximum": 4}, "debug": {"type": "boolean", "description": "Enable debug output"}, }, "required": ["prompt", "save_path", "task_note"],
• src/cli_agent_mcp/shared/invokers/banana.py:99-362 (helper)

  BananaInvoker.execute: Core execution logic, parameter processing, request building, API client invocation, and XML response formatting.
  class BananaInvoker: """Banana 图像生成调用器。 封装 Nano Banana Pro API，提供与 CLI Invoker 一致的接口。 Example: invoker = BananaInvoker() result = await invoker.execute(BananaParams( prompt="Create a cute cat", )) """ def __init__( self, event_callback: EventCallback | None = None, ) -> None: """初始化调用器。 Args: event_callback: 事件回调函数（用于 GUI 推送） """ self._event_callback = event_callback self._client: NanoBananaProClient | None = None @property def cli_type(self) -> str: return "banana" @property def cli_name(self) -> str: return "banana" def _get_client(self) -> NanoBananaProClient: """获取或创建 API 客户端。""" if self._client is None: self._client = NanoBananaProClient( event_callback=self._on_client_event, ) return self._client def _on_client_event(self, event: dict[str, Any]) -> None: """处理客户端事件并转发到 GUI。""" if not self._event_callback: return # 转换为 UnifiedEvent event_type = event.get("type", "") if event_type == "generation_started": unified = make_fallback_event( CLISource.UNKNOWN, { "type": "system", "subtype": "info", "message": f"Generating image: {event.get('prompt', '')[:50]}...", "source": "banana", }, ) elif event_type == "generation_completed": unified = make_fallback_event( CLISource.UNKNOWN, { "type": "system", "subtype": "info", "message": f"Generated {event.get('artifact_count', 0)} image(s)", "source": "banana", }, ) elif event_type == "generation_failed": unified = make_fallback_event( CLISource.UNKNOWN, { "type": "system", "subtype": "error", "severity": "error", "message": f"Generation failed: {event.get('error', 'Unknown error')}", "source": "banana", }, ) elif event_type == "api_retry": unified = make_fallback_event( CLISource.UNKNOWN, { "type": "system", "subtype": "warning", "severity": "warning", "message": f"API error {event.get('status_code')}, retrying in {event.get('delay')}s...", "source": "banana", }, ) else: return # 忽略其他事件 self._event_callback(unified) def _parse_images(self, images: list[dict[str, Any]]) -> list[ImageInput]: """解析图片输入列表。""" result = [] for img in images: source = img.get("source", "") if not source: continue source_path = Path(source) if not source_path.is_absolute(): logger.warning(f"Skipping non-absolute image path: {source}") continue if not source_path.exists(): logger.warning(f"Skipping non-existent image: {source_path}") continue # 解析角色 role_str = img.get("role", "") role = None if role_str: try: role = ImageRole(role_str) except ValueError: pass result.append(ImageInput( source=str(source_path), role=role, label=img.get("label", ""), )) return result def _build_response_xml(self, response: BananaResponse) -> str: """构建 XML 格式响应（不包含 base64）。""" request_id = html.escape(response.request_id, quote=True) model = html.escape(response.model, quote=True) # 构建 artifact_id -> path 映射 artifact_paths = {a.id: a.path for a in response.artifacts} lines = [ f'<nano-banana-response request_id="{request_id}" model="{model}">' ] # 按 candidate_index 分组 parts from itertools import groupby sorted_parts = sorted(response.parts, key=lambda p: p.candidate_index) for c_idx, group in groupby(sorted_parts, key=lambda p: p.candidate_index): lines.append(f' <candidate index="{c_idx}">') for part in group: if part.kind == "text": lines.append(f' <part index="{part.index}" kind="text">{_escape_xml(part.content)}</part>') elif part.kind == "thought": lines.append(f' <part index="{part.index}" kind="thought">{_escape_xml(part.content)}</part>') elif part.kind == "image": # 直接输出路径，减少 token 消耗 path = html.escape(artifact_paths.get(part.artifact_id, ""), quote=True) lines.append(f' <part index="{part.index}" kind="image" path="{path}"/>') lines.append(' </candidate>') # Grounding if response.grounding_html: lines.append(' <grounding>') lines.append(f' <html><![CDATA[{response.grounding_html}]]></html>') lines.append(' </grounding>') lines.append('</nano-banana-response>') return '\n'.join(lines) async def execute(self, params: BananaParams) -> BananaExecutionResult: """执行图像生成。 Args: params: 调用参数 Returns: 执行结果 """ start_time = time.time() # 验证参数 if not params.prompt: return BananaExecutionResult( success=False, error="prompt is required", ) # 处理 save_path（不再创建子目录） output_dir = params.save_path # 清理 task_note 用于文件名前缀 task_note = sanitize_task_note(params.task_note) # 解析宽高比 try: aspect_ratio = AspectRatio(params.aspect_ratio) except ValueError: aspect_ratio = AspectRatio.RATIO_1_1 # 解析分辨率 try: image_size = ImageSize(params.resolution) except ValueError: image_size = ImageSize.SIZE_1K # 构建请求 request = BananaRequest( prompt=params.prompt, images=self._parse_images(params.images), config=BananaConfig( aspect_ratio=aspect_ratio, image_size=image_size, use_search=params.use_search, include_thoughts=params.include_thoughts, temperature=params.temperature, top_p=params.top_p, top_k=params.top_k, num_images=params.num_images, ), output_dir=output_dir, task_note=task_note, ) # 执行生成 client = self._get_client() try: response = await client.generate(request) duration = time.time() - start_time if not response.success: return BananaExecutionResult( success=False, request_id=response.request_id, error=response.error, duration_sec=duration, model=response.model, api_endpoint=response.api_url, auth_token_masked=response.auth_hint, ) # 构建 XML 响应 response_xml = self._build_response_xml(response) return BananaExecutionResult( success=True, request_id=response.request_id, response_xml=response_xml, artifacts=[a.path for a in response.artifacts], duration_sec=duration, model=response.model, api_endpoint=response.api_url, auth_token_masked=response.auth_hint, ) except asyncio.CancelledError: # 取消错误必须 re-raise，不能被吞掉 raise except Exception as e: logger.exception(f"Banana execution failed: {e}") return BananaExecutionResult( success=False, error=str(e), duration_sec=time.time() - start_time, )
• src/cli_agent_mcp/shared/banana/client.py:59-516 (helper)

  NanoBananaProClient.generate: Performs the actual API call to Gemini image generation endpoint, handles retries, saves generated images to disk, parses response into structured BananaResponse.
  class NanoBananaProClient: """Nano Banana Pro API 客户端。 使用 aiohttp 异步调用 Gemini 3 Pro Image API。 Example: client = NanoBananaProClient() response = await client.generate(BananaRequest( prompt="Create a cute cat", config=BananaConfig(aspect_ratio=AspectRatio.RATIO_16_9), )) """ def __init__( self, config: BananaEnvConfig | None = None, event_callback: EventCallback | None = None, ) -> None: """初始化客户端。 Args: config: 环境配置（可选，默认从环境变量加载） event_callback: 事件回调函数（用于 GUI 推送） """ self._config = config or get_banana_config() self._event_callback = event_callback self._session: aiohttp.ClientSession | None = None async def _get_session(self) -> aiohttp.ClientSession: """获取或创建 HTTP 会话。""" if self._session is None or self._session.closed: self._session = aiohttp.ClientSession() return self._session async def close(self) -> None: """关闭 HTTP 会话。""" if self._session and not self._session.closed: await self._session.close() self._session = None def _emit_event(self, event: dict[str, Any]) -> None: """发送事件到回调。""" if self._event_callback: self._event_callback(event) def _mask_token(self, token: str) -> str: """脱敏 token，只显示前4位和后4位。""" if not token: return "(empty)" # 移除 Bearer 前缀 clean = token.replace("Bearer ", "") if len(clean) <= 8: return clean[:2] + "***" return f"{clean[:4]}...{clean[-4:]}" def _sanitize_headers(self, headers: dict[str, str]) -> dict[str, str]: """Sanitize headers for debug output, masking auth tokens.""" result = {} for k, v in headers.items(): if k.lower() in ("authorization", "x-goog-api-key"): result[k] = "***" else: result[k] = v return result def _build_image_metadata_prefix(self, images: list[ImageInput]) -> str: """Build metadata prefix from images with role/label.""" lines = [] for i, img in enumerate(images, 1): if img.role or img.label: role_str = img.role.value if img.role else "input" label_str = f' label="{img.label}"' if img.label else "" lines.append(f"Image {i}: role={role_str}{label_str}") return "\n".join(lines) + "\n\n" if lines else "" def _build_request_body(self, request: BananaRequest) -> dict[str, Any]: """构建 API 请求体。""" # 构建 contents parts: list[dict[str, Any]] = [] # 添加参考图片 for img in request.images: try: data, mime_type = encode_image_to_base64(img.source) parts.append({ "inline_data": { "mime_type": mime_type, "data": data, } }) except Exception as e: logger.warning(f"Failed to encode image {img.source}: {e}") # Build prompt with metadata prefix if images have role/label metadata_prefix = self._build_image_metadata_prefix(request.images) final_prompt = metadata_prefix + request.prompt if metadata_prefix else request.prompt # 添加文本提示词 parts.append({"text": final_prompt}) # 构建 generationConfig config = request.config generation_config: dict[str, Any] = { # 如果 use_search 或 include_thoughts，需要包含 TEXT "responseModalities": ["TEXT", "IMAGE"] if config.use_search or config.include_thoughts else ["IMAGE"], "temperature": config.temperature, "topP": config.top_p, "topK": config.top_k, } # 多图生成 if config.num_images > 1: generation_config["candidateCount"] = config.num_images # 图片配置 image_config: dict[str, Any] = {} if config.aspect_ratio: image_config["aspectRatio"] = config.aspect_ratio.value if config.image_size: image_config["imageSize"] = config.image_size.value if image_config: generation_config["imageConfig"] = image_config body: dict[str, Any] = { "contents": [{"parts": parts}], "generationConfig": generation_config, } # 思考配置（放在 generationConfig 内部） if config.include_thoughts: generation_config["thinkingConfig"] = {"includeThoughts": True} # 搜索工具 if config.use_search: body["tools"] = [{"google_search": {}}] return body async def _call_api( self, request: BananaRequest, request_id: str, ) -> tuple[dict[str, Any], str]: """调用 API 并处理重试。 Returns: (api_response, api_url) 元组 """ if not self._config.is_configured: raise BananaConfigError( "API token not configured. Set BANANA_AUTH_TOKEN or GOOGLE_API_KEY." ) url = f"{self._config.base_url}/models/{self._config.model}:generateContent" # 支持 Bearer token 和 API key 两种认证方式 auth_token = self._config.auth_token if auth_token.startswith("Bearer "): headers = { "Content-Type": "application/json", "Authorization": auth_token, } else: headers = { "Content-Type": "application/json", "x-goog-api-key": auth_token, } body = self._build_request_body(request) session = await self._get_session() for attempt in range(MAX_RETRIES): try: self._emit_event({ "type": "api_call", "request_id": request_id, "attempt": attempt + 1, "status": "started", }) # Emit api_request event with sanitized body self._emit_event({ "type": "api_request", "request_id": request_id, "url": url, "method": "POST", "headers": self._sanitize_headers(headers), "body": _sanitize_for_debug(body), }) start_time = time.time() async with session.post(url, json=body, headers=headers) as resp: duration_ms = int((time.time() - start_time) * 1000) resp_headers = dict(resp.headers) if resp.status == 200: api_response = await resp.json() # Emit api_response event with sanitized body self._emit_event({ "type": "api_response", "request_id": request_id, "status_code": resp.status, "duration_ms": duration_ms, "headers": resp_headers, "body": _sanitize_for_debug(api_response), }) return api_response, url error_text = await resp.text() # Emit api_response event for errors self._emit_event({ "type": "api_response", "request_id": request_id, "status_code": resp.status, "duration_ms": duration_ms, "headers": resp_headers, "body": error_text[:2000], }) # 可重试错误 if resp.status in (429, 500, 502, 503, 504): retry_after = resp.headers.get("Retry-After") delay = ( float(retry_after) if retry_after else RETRY_DELAYS[min(attempt, len(RETRY_DELAYS) - 1)] ) if attempt < MAX_RETRIES - 1: logger.warning( f"API error {resp.status}, retrying in {delay}s: {error_text[:200]}" ) self._emit_event({ "type": "api_retry", "request_id": request_id, "attempt": attempt + 1, "status_code": resp.status, "delay": delay, }) await asyncio.sleep(delay) continue raise BananaRetryableError(resp.status, error_text, delay) # 不可重试错误 raise BananaAPIError(resp.status, error_text) except aiohttp.ClientError as e: if attempt < MAX_RETRIES - 1: delay = RETRY_DELAYS[min(attempt, len(RETRY_DELAYS) - 1)] logger.warning(f"Network error, retrying in {delay}s: {e}") await asyncio.sleep(delay) continue raise BananaRetryableError(0, str(e)) # 不应该到达这里 raise BananaAPIError(0, "Max retries exceeded") def _find_next_seq(self, output_dir: Path, base_name: str, ext: str) -> int: """找到下一个可用的序号。""" seq = 0 while (output_dir / f"{base_name}_{seq}.{ext}").exists(): seq += 1 return seq def _save_image( self, data: bytes, output_dir: Path, task_note: str, mime_type: str, ) -> tuple[str, str]: """保存图片到文件。 Returns: (file_path, sha256) 元组 """ ext_map = {"image/png": "png", "image/jpeg": "jpeg", "image/webp": "webp"} ext = ext_map.get(mime_type, "png") output_dir.mkdir(parents=True, exist_ok=True) seq = self._find_next_seq(output_dir, task_note, ext) filename = f"{task_note}_{seq}.{ext}" file_path = output_dir / filename file_path.write_bytes(data) sha256 = hashlib.sha256(data).hexdigest() return str(file_path.absolute()), sha256 def _parse_response( self, api_response: dict[str, Any], request: BananaRequest, request_id: str, api_url: str = "", auth_hint: str = "", ) -> BananaResponse: """解析 API 响应。""" if not request.output_dir: raise BananaConfigError("output_dir is required") output_dir = Path(request.output_dir) # task_note 用于文件名前缀，如果为空则使用 request_id task_note = request.task_note or request_id parts: list[BananaPart] = [] artifacts: list[BananaArtifact] = [] grounding_html = "" candidates = api_response.get("candidates", []) for c_idx, candidate in enumerate(candidates): content = candidate.get("content", {}) for p_idx, part in enumerate(content.get("parts", [])): # 文本内容 if "text" in part: parts.append(BananaPart( index=p_idx, kind="text", content=part["text"], candidate_index=c_idx, )) # 思考内容 if "thought" in part: parts.append(BananaPart( index=p_idx, kind="thought", content=part["thought"], candidate_index=c_idx, )) # 图片内容（兼容 inlineData 和 inline_data） inline_data = part.get("inlineData") or part.get("inline_data") if inline_data: mime_type = inline_data.get("mimeType") or inline_data.get("mime_type", "image/png") b64_data = inline_data.get("data", "") if b64_data: image_bytes = base64.b64decode(b64_data) file_path, sha256 = self._save_image( image_bytes, output_dir, task_note, mime_type, ) artifact_id = f"img-{c_idx}-{p_idx}" artifacts.append(BananaArtifact( id=artifact_id, mime_type=mime_type, path=file_path, sha256=sha256, )) parts.append(BananaPart( index=p_idx, kind="image", artifact_id=artifact_id, candidate_index=c_idx, )) # Grounding 元数据 grounding_meta = candidate.get("groundingMetadata", {}) search_entry = grounding_meta.get("searchEntryPoint", {}) if "renderedContent" in search_entry: grounding_html = search_entry["renderedContent"] return BananaResponse( request_id=request_id, model=self._config.model, parts=parts, artifacts=artifacts, grounding_html=grounding_html, success=True, api_url=api_url, auth_hint=auth_hint, ) async def generate(self, request: BananaRequest) -> BananaResponse: """生成图片。 Args: request: 生成请求 Returns: BananaResponse 响应对象 """ request_id = str(uuid.uuid4())[:8] # 预先构建 debug 信息 api_url = f"{self._config.base_url}/models/{self._config.model}:generateContent" auth_hint = self._mask_token(self._config.auth_token) self._emit_event({ "type": "generation_started", "request_id": request_id, "prompt": request.prompt[:100], "image_count": len(request.images), }) try: api_response, api_url = await self._call_api(request, request_id) response = self._parse_response(api_response, request, request_id, api_url, auth_hint) self._emit_event({ "type": "generation_completed", "request_id": request_id, "artifact_count": len(response.artifacts), "success": True, }) return response except (BananaAPIError, BananaRetryableError, BananaConfigError) as e: self._emit_event({ "type": "generation_failed", "request_id": request_id, "error": str(e), "api_url": api_url, "auth_hint": auth_hint, }) return BananaResponse( request_id=request_id, model=self._config.model, success=False, error=str(e), api_url=api_url, auth_hint=auth_hint, ) except asyncio.CancelledError: # 取消错误必须 re-raise，不能被吞掉 self._emit_event({ "type": "generation_cancelled", "request_id": request_id, }) raise except Exception as e: logger.exception(f"Unexpected error in generate: {e}") self._emit_event({ "type": "generation_failed", "request_id": request_id, "error": str(e), "api_url": api_url, "auth_hint": auth_hint, }) return BananaResponse( request_id=request_id, model=self._config.model, success=False, error=f"Unexpected error: {e}", api_url=api_url, auth_hint=auth_hint, )