Remove MiniMax MCP support

.env.example

@@ -2,4 +2,4 @@ TG_API_ID=123456
 TG_API_HASH=your_api_hash_here
 SESSION_DIR=sessions
 ACCOUNTS_JSON=[{"name":"personal","session":"personal"},{"name":"work","session":"work"}]
-# AI_CONFIG={"default_model":"mm","context_messages":12,"prompt":{"root":"prompts/ai","system_file":"system.md","parts_dir":"parts"},"image_understand":{"api_key":"your_minimax_token_plan_key_here","api_host":"https://api.minimax.chat","command":["uvx","minimax-coding-plan-mcp","-y"]},"providers":{"minimax":{"api_key":"your_minimax_api_key_here","base_url":"https://api.minimax.io/v1"},"openrouter":{"api_key":"your_openrouter_api_key_here","base_url":"https://openrouter.ai/api/v1"}},"models":{"mm":{"provider":"minimax","model":"MiniMax-Text-01"},"or":{"provider":"openrouter","model":"openai/gpt-4o-mini"}}}
+# AI_CONFIG={"default_model":"mm","context_messages":12,"prompt":{"root":"prompts/ai","system_file":"system.md","parts_dir":"parts"},"providers":{"minimax":{"api_key":"your_minimax_api_key_here","base_url":"https://api.minimax.io/v1"},"openrouter":{"api_key":"your_openrouter_api_key_here","base_url":"https://openrouter.ai/api/v1"}},"models":{"mm":{"provider":"minimax","model":"MiniMax-Text-01"},"or":{"provider":"openrouter","model":"openai/gpt-4o-mini"}}}

CLAUDE.md

@@ -13,16 +13,14 @@ Logs go to `bot.log` by default (rotating, 10 MB, 3 backups). Override with `LOG
 
 ## Configuration
 
-Copy `config.example.json` to `config.json` and fill in credentials, or use environment variables (`.env.example` shows the env-var form). Both sources are merged at startup — env vars take precedence.
+Copy `config.example.json` to `config.json` and fill in credentials, or use environment variables (`.env.example` shows the env-var form). Both sources are merged at startup; env vars take precedence.
 
 Key env vars: `TG_API_ID`, `TG_API_HASH`, `SESSION_DIR`, `ACCOUNTS_JSON` (JSON list) or `ACCOUNT_SESSIONS` (comma-separated session names), `AI_CONFIG` (full AI block as JSON), `BOT_CONFIG` (path to config file), `BOT_DOTENV_PATH` (path to .env).
 
-`ai.image_understand.command` defaults to `["uvx", "minimax-coding-plan-mcp", "-y"]`. If `uvx` is not on PATH but `uv` is available via Python, use `["python3", "-m", "uv", "tool", "run", "minimax-coding-plan-mcp", "-y"]` instead.
-
 ## Architecture
 
 ### Boot path
-`main.py` → `load_app_config()` → `AccountManager` → one Pyrogram (`kurigram`) `Client` per account → `register_common_handlers()` → `pyrogram.idle()`.
+`main.py` -> `load_app_config()` -> `AccountManager` -> one Pyrogram (`kurigram`) `Client` per account -> `register_common_handlers()` -> `pyrogram.idle()`.
 
 ### Command system (`app/commands/`)
 - **Dispatcher** (`dispatcher.py`): parses `.command` (owner prefix) or `..command` (shared/guest prefix) from message text, resolves to a `CommandDefinition`, runs pre-checks, calls handler.
@@ -32,15 +30,14 @@ Key env vars: `TG_API_ID`, `TG_API_HASH`, `SESSION_DIR`, `ACCOUNTS_JSON` (JSON l
 - **ACL** (`acl.py`): `RuntimeAclStore` persists per-session `<session>.acl.json` allow/deny rules. Only shared (`.is_shared`) commands can be granted to guests.
 
 ### AI system (`app/ai/`)
-- **AIService** (`service.py`): OpenAI-compatible client (via `openai` SDK + `httpx`) that supports multiple named providers and model aliases defined in config. History collection reads `context_messages` prior messages from the chat. Reactions (🤔/👍/👎) are used for status signaling.
+- **AIService** (`service.py`): OpenAI-compatible client (via `openai` SDK + `httpx`) that supports multiple named providers and model aliases defined in config. History collection reads `context_messages` prior messages from the chat. Reactions are used for status signaling.
-- **Parser** (`parser.py`): `.ai [N] [--img] [--model-alias] prompt` — leading integer overrides context window, `--img` enables image understanding, other `--flags` are treated as model aliases.
+- **Parser** (`parser.py`): `.ai [N] [--model-alias] prompt` - leading integer overrides context window, other `--flags` are treated as model aliases.
 - **Prompts** (`prompts.py`): `PromptBook` loads `prompts/ai/system.md` + all `prompts/ai/parts/*.md` (sorted), joined with double newline. Edit those files to tune AI behavior without touching code.
-- **Image understanding** (`image_understand.py`): Spawns the MiniMax MCP subprocess (`uvx minimax-coding-plan-mcp`) and communicates via JSON-RPC over stdio (LSP-style `Content-Length` framing). Images are downloaded to a temp dir, analyzed one by one, and injected as a system message before the final completion.
 
 ### Key types
-- `CommandContext` — per-invocation context passed to every handler; holds `client`, `manager`, `account`, `message`, parsed `args`/`argv`, and convenience properties (`actor_id`, `owner_id`, `is_owner`, `replied_user`).
+- `CommandContext` - per-invocation context passed to every handler; holds `client`, `manager`, `account`, `message`, parsed `args`/`argv`, and convenience properties (`actor_id`, `owner_id`, `is_owner`, `replied_user`).
-- `AccountState` — runtime state per account (client, ACL store, status, `me` object).
+- `AccountState` - runtime state per account (client, ACL store, status, `me` object).
-- `AIInvocation` — parsed AI call parameters (context_messages, model_flags, use_image_understanding, prompt).
+- `AIInvocation` - parsed AI call parameters (context_messages, model_flags, prompt).
 
 ## Adding a new command
 

app/ai/image_understand.py

@@ -1,407 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from collections import deque
-from contextlib import suppress
-import json
-import os
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any
-
-from app.config import AIImageUnderstandConfig
-
-
-SUPPORTED_IMAGE_MIME_PREFIXES = ("image/",)
-SUPPORTED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
-MAX_IMAGE_BYTES = 20 * 1024 * 1024
-
-
-@dataclass(slots=True)
-class TelegramImageSource:
-    message_id: int
-    index: int
-    kind: str
-    media: Any
-    unique_id: str
-    mime_type: str | None = None
-    file_name: str | None = None
-    text_context: str = ""
-
-    @property
-    def label(self) -> str:
-        return f"message {self.message_id} #{self.index + 1} ({self.kind})"
-
-
-def _text_context(message: Any) -> str:
-    text = getattr(message, "text", None) or getattr(message, "caption", None)
-    if isinstance(text, str):
-        return text.strip()
-    return ""
-
-
-def _is_supported_image(media: Any) -> bool:
-    mime_type = getattr(media, "mime_type", None)
-    if isinstance(mime_type, str) and mime_type.startswith(SUPPORTED_IMAGE_MIME_PREFIXES):
-        return True
-
-    file_name = getattr(media, "file_name", None)
-    if isinstance(file_name, str):
-        suffix = Path(file_name).suffix.lower()
-        if suffix in SUPPORTED_IMAGE_EXTENSIONS:
-            return True
-
-    return False
-
-
-def extract_image_sources(message: Any) -> list[TelegramImageSource]:
-    message_id = int(getattr(message, "id", 0) or 0)
-    text_context = _text_context(message)
-    sources: list[TelegramImageSource] = []
-
-    candidates: list[tuple[str, Any]] = []
-    photo = getattr(message, "photo", None)
-    if photo is not None:
-        candidates.append(("photo", photo))
-
-    document = getattr(message, "document", None)
-    if document is not None and _is_supported_image(document):
-        candidates.append(("document", document))
-
-    animation = getattr(message, "animation", None)
-    if animation is not None and _is_supported_image(animation):
-        candidates.append(("animation", animation))
-
-    sticker = getattr(message, "sticker", None)
-    if sticker is not None and _is_supported_image(sticker):
-        candidates.append(("sticker", sticker))
-
-    for index, (kind, media) in enumerate(candidates):
-        unique_id = (
-            str(getattr(media, "file_unique_id", "") or "")
-            or str(getattr(media, "file_id", "") or "")
-            or f"{message_id}:{kind}:{index}"
-        )
-        sources.append(
-            TelegramImageSource(
-                message_id=message_id,
-                index=index,
-                kind=kind,
-                media=media,
-                unique_id=unique_id,
-                mime_type=getattr(media, "mime_type", None),
-                file_name=getattr(media, "file_name", None),
-                text_context=text_context,
-            )
-        )
-
-    return sources
-
-
-class MCPError(RuntimeError):
-    pass
-
-
-class MinimaxImageUnderstandClient:
-    def __init__(self, config: AIImageUnderstandConfig) -> None:
-        self.config = config
-        self._process: asyncio.subprocess.Process | None = None
-        self._request_id = 0
-        self._stdout: asyncio.StreamReader | None = None
-        self._stdin: asyncio.StreamWriter | None = None
-        self._stderr: asyncio.StreamReader | None = None
-        self._stderr_task: asyncio.Task[None] | None = None
-        self._stderr_tail: deque[str] = deque(maxlen=20)
-        self._lock = asyncio.Lock()
-
-    async def __aenter__(self) -> "MinimaxImageUnderstandClient":
-        await self.start()
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb) -> None:
-        await self.close()
-
-    async def start(self) -> None:
-        if self._process is not None:
-            return
-
-        env = os.environ.copy()
-        env["MINIMAX_API_KEY"] = self.config.api_key
-        env["MINIMAX_API_HOST"] = self.config.api_host
-
-        try:
-            self._process = await asyncio.create_subprocess_exec(
-                *self.config.command,
-                stdin=asyncio.subprocess.PIPE,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-                env=env,
-            )
-        except FileNotFoundError as exc:
-            raise MCPError(
-                f"Unable to start MiniMax MCP command: {self.config.command[0]}"
-            ) from exc
-
-        if self._process.stdin is None or self._process.stdout is None:
-            await self._terminate_process()
-            raise MCPError("MiniMax MCP process did not expose stdio pipes")
-
-        self._stdin = self._process.stdin
-        self._stdout = self._process.stdout
-        self._stderr = self._process.stderr
-        self._stderr_task = asyncio.create_task(self._drain_stderr())
-        request_timeout = self.config.timeout if self.config.timeout is not None else 120.0
-
-        try:
-            await self._request(
-                "initialize",
-                {
-                    "protocolVersion": "2024-11-05",
-                    "capabilities": {},
-                    "clientInfo": {
-                        "name": "multi-bot-tg",
-                        "version": "1.0.0",
-                    },
-                },
-                timeout=request_timeout,
-            )
-            await self._notify("notifications/initialized", {})
-        except Exception:
-            await self._terminate_process()
-            raise
-
-    async def close(self) -> None:
-        process = self._process
-        self._process = None
-
-        if process is None:
-            return
-
-        try:
-            if self._stdin is not None:
-                try:
-                    await asyncio.wait_for(self._request("shutdown", {}, timeout=None), timeout=5)
-                except Exception:
-                    pass
-        finally:
-            try:
-                process.terminate()
-            except ProcessLookupError:
-                pass
-            try:
-                await asyncio.wait_for(process.wait(), timeout=5)
-            except Exception:
-                try:
-                    process.kill()
-                except ProcessLookupError:
-                    pass
-
-        self._stdin = None
-        self._stdout = None
-        self._stderr = None
-        await self._stop_stderr_task()
-
-    async def list_tools(self) -> list[dict]:
-        if self._process is None:
-            await self.start()
-        timeout = self.config.timeout if self.config.timeout is not None else 30.0
-        result = await self._request("tools/list", {}, timeout=timeout)
-        if isinstance(result, dict):
-            tools = result.get("tools")
-            if isinstance(tools, list):
-                return tools
-        return []
-
-    async def call_tool(self, name: str, arguments: dict) -> str:
-        if self._process is None:
-            await self.start()
-        timeout = self.config.timeout if self.config.timeout is not None else 120.0
-        result = await self._request(
-            "tools/call",
-            {"name": name, "arguments": arguments},
-            timeout=timeout,
-        )
-        text = self._extract_text(result)
-        if not text:
-            raise MCPError(f"MCP tool '{name}' returned an empty response")
-        return text
-
-    async def understand_image(self, *, image_path: Path, prompt: str) -> str:
-        if self._process is None:
-            await self.start()
-
-        size = image_path.stat().st_size
-        if size > MAX_IMAGE_BYTES:
-            raise MCPError(
-                f"Image {image_path.name} is too large for understand_image ({size} bytes)"
-            )
-        request_timeout = self.config.timeout if self.config.timeout is not None else 120.0
-
-        last_error: MCPError | None = None
-        for image_field in ("image_url", "image_source"):
-            try:
-                result = await self._request(
-                    "tools/call",
-                    {
-                        "name": "understand_image",
-                        "arguments": {
-                            "prompt": prompt,
-                            image_field: str(image_path),
-                        },
-                    },
-                    timeout=request_timeout,
-                )
-                break
-            except MCPError as exc:
-                last_error = exc
-        else:
-            raise last_error or MCPError("MiniMax understand_image call failed")
-
-        text = self._extract_text(result)
-        if not text:
-            raise MCPError("MiniMax understand_image returned an empty response")
-        return text
-
-    async def _notify(self, method: str, params: dict[str, Any]) -> None:
-        await self._write_message({"jsonrpc": "2.0", "method": method, "params": params})
-
-    async def _request(self, method: str, params: dict[str, Any], *, timeout: float | None = None) -> Any:
-        async with self._lock:
-            self._request_id += 1
-            request_id = self._request_id
-            await self._write_message(
-                {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}
-            )
-
-            async def _wait_for_response() -> Any:
-                while True:
-                    message = await self._read_message()
-                    if not isinstance(message, dict):
-                        continue
-                    if message.get("id") != request_id:
-                        continue
-                    if "error" in message:
-                        error = message["error"]
-                        message_text = error.get("message") if isinstance(error, dict) else None
-                        raise MCPError(message_text or f"MCP request failed for {method}")
-                    return message.get("result")
-
-            if timeout is None:
-                return await _wait_for_response()
-            try:
-                return await asyncio.wait_for(_wait_for_response(), timeout=timeout)
-            except TimeoutError as exc:
-                raise MCPError(
-                    f"Timed out waiting for MiniMax MCP '{method}' response after {timeout:.0f}s"
-                    f"{self._format_stderr_tail()}"
-                ) from exc
-
-    async def _write_message(self, payload: dict[str, Any]) -> None:
-        if self._stdin is None:
-            raise MCPError("MiniMax MCP stdin is not available")
-
-        data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-        header = f"Content-Length: {len(data)}\r\n\r\n".encode("ascii")
-        self._stdin.write(header + data)
-        await self._stdin.drain()
-
-    async def _read_message(self) -> Any:
-        if self._stdout is None:
-            raise MCPError("MiniMax MCP stdout is not available")
-
-        header_block = await self._read_header_block()
-        headers = header_block.decode("ascii", errors="ignore").split("\r\n")
-        content_length = None
-        for header in headers:
-            header = header.strip()
-            if header.lower().startswith("content-length:"):
-                content_length = int(header.split(":", 1)[1].strip())
-                break
-        if content_length is None:
-            raise MCPError("MiniMax MCP response is missing Content-Length header")
-
-        payload = await self._stdout.readexactly(content_length)
-        return json.loads(payload.decode("utf-8"))
-
-    async def _read_header_block(self) -> bytes:
-        if self._stdout is None:
-            raise MCPError("MiniMax MCP stdout is not available")
-
-        buffer = bytearray()
-        while True:
-            chunk = await self._stdout.read(1)
-            if not chunk:
-                raise MCPError(f"MiniMax MCP stdout closed before a response{self._format_stderr_tail()}")
-            buffer.extend(chunk)
-            if buffer.endswith(b"\r\n\r\n"):
-                return bytes(buffer)
-            if buffer.endswith(b"\n\n"):
-                return bytes(buffer).replace(b"\n", b"\r\n")
-            if len(buffer) > 64 * 1024:
-                raise MCPError("MiniMax MCP response headers exceeded 64KB")
-
-    async def _drain_stderr(self) -> None:
-        if self._stderr is None:
-            return
-        while True:
-            line = await self._stderr.readline()
-            if not line:
-                return
-            text = line.decode("utf-8", errors="replace").strip()
-            if text:
-                self._stderr_tail.append(text)
-
-    def _format_stderr_tail(self) -> str:
-        if not self._stderr_tail:
-            return ""
-        return "\nMiniMax MCP stderr tail:\n" + "\n".join(self._stderr_tail)
-
-    async def _stop_stderr_task(self) -> None:
-        task = self._stderr_task
-        self._stderr_task = None
-        if task is None:
-            return
-        task.cancel()
-        with suppress(asyncio.CancelledError):
-            await task
-
-    async def _terminate_process(self) -> None:
-        process = self._process
-        self._process = None
-        if process is None:
-            return
-
-        try:
-            process.terminate()
-        except ProcessLookupError:
-            pass
-        try:
-            await asyncio.wait_for(process.wait(), timeout=5)
-        except Exception:
-            try:
-                process.kill()
-            except ProcessLookupError:
-                pass
-        await self._stop_stderr_task()
-        self._stdin = None
-        self._stdout = None
-        self._stderr = None
-
-    def _extract_text(self, result: Any) -> str:
-        if isinstance(result, dict):
-            content = result.get("content")
-            if isinstance(content, list):
-                parts: list[str] = []
-                for item in content:
-                    if isinstance(item, dict):
-                        text = item.get("text")
-                        if isinstance(text, str):
-                            parts.append(text.strip())
-                return "\n".join(part for part in parts if part).strip()
-            if isinstance(content, str):
-                return content.strip()
-            if isinstance(result.get("text"), str):
-                return str(result["text"]).strip()
-        if isinstance(result, str):
-            return result.strip()
-        return ""

app/ai/parser.py

@@ -8,7 +8,6 @@ import shlex
 class AIInvocation:
     context_messages: int
     model_flags: list[str]
-    use_image_understanding: bool
     prompt: str
 
 
@@ -20,7 +19,6 @@ def parse_ai_invocation(text: str, *, default_context_messages: int) -> AIInvoca
 
     context_messages = default_context_messages
     model_flags: list[str] = []
-    use_image_understanding = False
 
     if tokens:
         first = tokens[0]
@@ -33,18 +31,14 @@ def parse_ai_invocation(text: str, *, default_context_messages: int) -> AIInvoca
             flag = flag.split("=", 1)[0]
         flag = flag.strip().lower()
         if flag:
-            if flag == "img":
+            model_flags.append(flag)
-                use_image_understanding = True
-            else:
-                model_flags.append(flag)
 
     prompt = " ".join(tokens).strip()
     if not prompt:
-        raise ValueError("Usage: ..ai [N] [--img] [--model] prompt")
+        raise ValueError("Usage: ..ai [N] [--model] prompt")
 
     return AIInvocation(
         context_messages=context_messages,
         model_flags=model_flags,
-        use_image_understanding=use_image_understanding,
         prompt=prompt,
     )

app/ai/service.py

@@ -7,21 +7,14 @@ import json
 import logging
 from pathlib import Path
 import sys
-import tempfile
 from typing import TYPE_CHECKING, Any, Awaitable, Callable
 
 import httpx
 from openai import AsyncOpenAI
 
-from app.ai.image_understand import (
-    MCPError,
-    MinimaxImageUnderstandClient,
-    TelegramImageSource,
-    extract_image_sources,
-)
 from app.ai.parser import AIInvocation
 from app.ai.prompts import PromptBook
-from app.ai.tools import FILE_TOOL_DISPATCH, FILE_TOOL_SCHEMAS, PYTHON_EXEC_SCHEMA, mcp_tools_to_openai_schemas
+from app.ai.tools import FILE_TOOL_DISPATCH, FILE_TOOL_SCHEMAS, PYTHON_EXEC_SCHEMA
 from app.config import AIConfig, AIModelConfig, AIProviderConfig
 
 if TYPE_CHECKING:
@@ -61,7 +54,6 @@ class HistoryMessage:
     message: Any
     role: str
     text: str
-    images: list[TelegramImageSource]
 
 
 class AIService:
@@ -70,11 +62,9 @@ class AIService:
         self.prompt_book = PromptBook(config.prompt.root)
         self._clients: dict[str, AsyncOpenAI] = {}
         self._system_prompt: str | None = None
-        self._mcp_tool_schemas: list[dict] | None = None
-        self._mcp_tools_fetched: bool = False
 
     async def initialize(self) -> None:
-        await self._fetch_mcp_tool_schemas()
+        return None
 
     @property
     def enabled(self) -> bool:
@@ -89,8 +79,6 @@ class AIService:
             model = self.config.models[alias]
             provider = self.config.providers[model.provider]
             lines.append(f"{alias} -> {provider.name}:{model.model}")
-        if self.config.image_understand is not None:
-            lines.append("image-understand=enabled")
         return "; ".join(lines)
 
     def select_model_alias(self, model_flags: list[str]) -> str:
@@ -173,122 +161,13 @@ class AIService:
 
             text = self._extract_message_text(message)
             role = "assistant" if self._is_own_message(message, owner_id) else "user"
-            images = extract_image_sources(message)
+            collected.append(HistoryMessage(message=message, role=role, text=text))
-            collected.append(HistoryMessage(message=message, role=role, text=text, images=images))
             if len(collected) >= limit:
                 break
 
         collected.reverse()
         return collected
 
-    async def _download_image_source(
-        self,
-        ctx: "CommandContext",
-        source: TelegramImageSource,
-        directory: Path,
-    ) -> Path:
-        path = await ctx.client.download_media(source.media, file_name=str(directory))
-        if not isinstance(path, str) or not path:
-            raise RuntimeError(f"Failed to download {source.label}")
-        downloaded = Path(path)
-        if not downloaded.exists():
-            raise RuntimeError(f"Downloaded file for {source.label} does not exist")
-        return downloaded
-
-    def _build_image_prompt(self, invocation: AIInvocation, source: TelegramImageSource) -> str:
-        context_text = source.text_context or "none"
-        return "\n".join(
-            [
-                "Analyze the attached Telegram image for a user-bot.",
-                f"User request: {invocation.prompt}",
-                f"Source: {source.label}",
-                f"Message text or caption: {context_text}",
-                "Return a concise structured description of visible objects, readable text, notable actions, and anything relevant to the user request.",
-            ]
-        )
-
-    async def _analyze_images(
-        self,
-        ctx: "CommandContext",
-        invocation: AIInvocation,
-        history: list[HistoryMessage],
-        *,
-        progress: ProgressCallback | None,
-    ) -> list[str]:
-        if self.config.image_understand is None:
-            raise ValueError("Image understanding is not configured")
-
-        sources: list[TelegramImageSource] = []
-        seen: set[str] = set()
-        for item in history:
-            for source in item.images:
-                if source.unique_id in seen:
-                    continue
-                seen.add(source.unique_id)
-                sources.append(source)
-
-        if not sources:
-            await self._maybe_call_progress(
-                progress,
-                "🛈 No images found in the selected context; continuing with text only",
-            )
-            return []
-
-        analyses: list[str] = []
-        await self._maybe_call_progress(
-            progress,
-            f"🛠 Found {len(sources)} image(s) in context, running understand_image",
-        )
-
-        with tempfile.TemporaryDirectory(prefix="multi-bot-tg-ai-") as tmpdir:
-            temp_root = Path(tmpdir)
-            try:
-                async with MinimaxImageUnderstandClient(self.config.image_understand) as image_client:
-                    for index, source in enumerate(sources, start=1):
-                        await self._maybe_call_progress(
-                            progress,
-                            f"🛠 Understanding image {index}/{len(sources)} from message {source.message_id}",
-                        )
-                        downloaded = await self._download_image_source(ctx, source, temp_root)
-                        if downloaded.stat().st_size > 20 * 1024 * 1024:
-                            raise RuntimeError(
-                                f"Image {source.label} exceeds the 20MB understand_image limit"
-                            )
-
-                        try:
-                            text = await image_client.understand_image(
-                                image_path=downloaded,
-                                prompt=self._build_image_prompt(invocation, source),
-                            )
-                        except MCPError as exc:
-                            raise RuntimeError(
-                                f"Image analysis failed for {source.label}: {exc}"
-                            ) from exc
-
-                        analyses.append(
-                            "\n".join(
-                                [
-                                    f"[Image {index}/{len(sources)}]",
-                                    f"Source: {source.label}",
-                                    f"File: {downloaded.name}",
-                                    f"Analysis:\n{text}",
-                                ]
-                            )
-                        )
-            except MCPError as exc:
-                raise RuntimeError(f"Image understanding tool failed: {exc}") from exc
-
-        await self._maybe_call_progress(progress, "🛠 Image analysis complete")
-        return analyses
-
-    def _build_image_summary_message(self, image_notes: list[str]) -> str:
-        return "\n\n".join(
-            [
-                "Visual context extracted from images:",
-                *image_notes,
-            ]
-        ).strip()
-
     def _render_user_message(self, text: str, user: Any) -> str:
         template = self.config.prompt.user_template
         if not template or not text:
@@ -316,11 +195,8 @@ class AIService:
         *,
         prompt: str,
         history: list[HistoryMessage],
-        image_notes: list[str],
     ) -> list[dict[str, Any]]:
         messages: list[dict[str, Any]] = [{"role": "system", "content": self.load_system_prompt()}]
-        if image_notes:
-            messages.append({"role": "system", "content": self._build_image_summary_message(image_notes)})
         for item in history:
             if not item.text:
                 continue
@@ -333,30 +209,8 @@ class AIService:
         messages.append({"role": "user", "content": self._render_user_message(prompt, ctx.actor)})
         return messages
 
-    async def _fetch_mcp_tool_schemas(self) -> list[dict]:
-        if self._mcp_tools_fetched:
-            return self._mcp_tool_schemas or []
-        if self.config.image_understand is None:
-            log.info("MCP tools: image_understand not configured, skipping")
-            self._mcp_tool_schemas = []
-            self._mcp_tools_fetched = True
-            return []
-        log.info("MCP tools: fetching tool list from %s", self.config.image_understand.command)
-        try:
-            async with MinimaxImageUnderstandClient(self.config.image_understand) as mcp_client:
-                raw = await mcp_client.list_tools()
-            self._mcp_tool_schemas = mcp_tools_to_openai_schemas(raw)
-            self._mcp_tools_fetched = True
-            names = [s["function"]["name"] for s in self._mcp_tool_schemas]
-            log.info("MCP tools: loaded %d tool(s): %s", len(names), ", ".join(names) or "(none)")
-        except Exception:
-            log.warning("MCP tools: failed to load, MiniMax tools will be unavailable", exc_info=True)
-            self._mcp_tool_schemas = []
-        return self._mcp_tool_schemas or []
-
     async def _build_tool_list(self) -> list[dict]:
-        mcp_schemas = await self._fetch_mcp_tool_schemas()
+        return FILE_TOOL_SCHEMAS + [PYTHON_EXEC_SCHEMA]
-        return FILE_TOOL_SCHEMAS + [PYTHON_EXEC_SCHEMA] + mcp_schemas
 
     def _get_allowed_roots(self) -> tuple[Path, ...]:
         # [0] = base for resolution (project root); [1:] = permitted subdirs
@@ -471,8 +325,6 @@ class AIService:
         self,
         tool_call: Any,
         ctx: "CommandContext",
-        mcp_client: MinimaxImageUnderstandClient | None,
-        mcp_names: set[str],
         allowed_roots: tuple[Path, ...],
         progress: ProgressCallback | None,
     ) -> str:
@@ -491,14 +343,10 @@ class AIService:
                 result = await self._execute_send_file(arguments, ctx, allowed_roots)
             elif name in FILE_TOOL_DISPATCH:
                 result = await FILE_TOOL_DISPATCH[name](arguments, allowed_roots)
-            elif name in mcp_names and mcp_client is not None:
-                result = await mcp_client.call_tool(name, arguments)
             else:
                 result = f"Unknown tool: {name!r}"
         except (ValueError, OSError) as exc:
             result = f"Tool {name!r} error: {exc}"
-        except MCPError as exc:
-            result = f"Tool {name!r} MCP error: {exc}"
 
         await self._maybe_call_progress(progress, f"🛠 Tool {name}: {result[:80]}")
         return result
@@ -514,108 +362,79 @@ class AIService:
         client = self._get_client(model.provider)
 
         history = await self.collect_history(ctx, limit=invocation.context_messages)
-        image_notes: list[str] = []
-
-        if invocation.use_image_understanding:
-            image_notes = await self._analyze_images(
-                ctx,
-                invocation,
-                history,
-                progress=progress,
-            )
 
         await self._maybe_call_progress(progress, "🛠 Generating final answer")
         messages = await self.build_messages(
             ctx,
             prompt=invocation.prompt,
             history=history,
-            image_notes=image_notes,
         )
 
         tools = await self._build_tool_list()
         allowed_roots = self._get_allowed_roots()
         max_rounds = self.config.tool_max_rounds
-        mcp_names: set[str] = {
-            s["function"]["name"]
-            for s in (self._mcp_tool_schemas or [])
-        }
 
-        mcp_client: MinimaxImageUnderstandClient | None = None
+        for round_num in range(max_rounds + 1):
+            kwargs: dict[str, Any] = {
+                "model": model.model.model,
+                "messages": messages,
+            }
+            if tools:
+                kwargs["tools"] = tools
 
-        try:
+            try:
-            for round_num in range(max_rounds + 1):
+                response = await client.chat.completions.create(**kwargs)
-                kwargs: dict[str, Any] = {
+            except Exception as exc:
-                    "model": model.model.model,
+                log.exception("AI request failed for model %s", model.alias)
-                    "messages": messages,
+                raise RuntimeError(f"AI request failed: {exc}") from exc
-                }
-                if tools:
-                    kwargs["tools"] = tools
 
-                try:
+            choice = response.choices[0]
-                    response = await client.chat.completions.create(**kwargs)
+            finish_reason = getattr(choice, "finish_reason", None)
-                except Exception as exc:
-                    log.exception("AI request failed for model %s", model.alias)
-                    raise RuntimeError(f"AI request failed: {exc}") from exc
 
-                choice = response.choices[0]
+            if finish_reason != "tool_calls":
-                finish_reason = getattr(choice, "finish_reason", None)
+                text = self._extract_response_text(response)
+                if not text:
+                    raise RuntimeError("AI model returned an empty response")
+                return AIResponse(text=text, model=model.model, provider=model.provider)
 
-                if finish_reason != "tool_calls":
+            if round_num >= max_rounds:
-                    text = self._extract_response_text(response)
+                raise RuntimeError(
-                    if not text:
+                    f"Agentic loop exceeded the maximum of {max_rounds} tool rounds"
-                        raise RuntimeError("AI model returned an empty response")
+                )
-                    return AIResponse(text=text, model=model.model, provider=model.provider)
 
-                if round_num >= max_rounds:
+            tool_calls = getattr(choice.message, "tool_calls", None) or []
-                    raise RuntimeError(
+            if not tool_calls:
-                        f"Agentic loop exceeded the maximum of {max_rounds} tool rounds"
+                raise RuntimeError("finish_reason was tool_calls but no tool_calls present")
-                    )
 
-                tool_calls = getattr(choice.message, "tool_calls", None) or []
+            messages.append({
-                if not tool_calls:
+                "role": "assistant",
-                    raise RuntimeError("finish_reason was tool_calls but no tool_calls present")
+                "content": getattr(choice.message, "content", None),
+                "tool_calls": [
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {
+                            "name": tc.function.name,
+                            "arguments": tc.function.arguments,
+                        },
+                    }
+                    for tc in tool_calls
+                ],
+            })
 
+            for tc in tool_calls:
+                tc_name = getattr(getattr(tc, "function", None), "name", "?")
+                await self._maybe_call_progress(progress, f"🛠 Calling tool: {tc_name}")
+                result_text = await self._execute_tool_call(
+                    tc, ctx, allowed_roots, progress
+                )
                 messages.append({
-                    "role": "assistant",
+                    "role": "tool",
-                    "content": getattr(choice.message, "content", None),
+                    "tool_call_id": tc.id,
-                    "tool_calls": [
+                    "content": result_text,
-                        {
-                            "id": tc.id,
-                            "type": "function",
-                            "function": {
-                                "name": tc.function.name,
-                                "arguments": tc.function.arguments,
-                            },
-                        }
-                        for tc in tool_calls
-                    ],
                 })
 
-                needs_mcp = any(
+        raise RuntimeError("Agentic loop terminated without a response")
-                    getattr(getattr(tc, "function", None), "name", "") in mcp_names
-                    for tc in tool_calls
-                )
-                if needs_mcp and mcp_client is None and self.config.image_understand is not None:
-                    mcp_client = MinimaxImageUnderstandClient(self.config.image_understand)
-                    await mcp_client.start()
-
-                for tc in tool_calls:
-                    tc_name = getattr(getattr(tc, "function", None), "name", "?")
-                    await self._maybe_call_progress(progress, f"🛠 Calling tool: {tc_name}")
-                    result_text = await self._execute_tool_call(
-                        tc, ctx, mcp_client, mcp_names, allowed_roots, progress
-                    )
-                    messages.append({
-                        "role": "tool",
-                        "tool_call_id": tc.id,
-                        "content": result_text,
-                    })
-
-            raise RuntimeError("Agentic loop terminated without a response")
-
-        finally:
-            if mcp_client is not None:
-                await mcp_client.close()
 
     def _extract_response_text(self, response: Any) -> str:
         choices = getattr(response, "choices", None) or []

app/ai/tools.py

@@ -373,28 +373,3 @@ FILE_TOOL_SCHEMAS: list[dict[str, Any]] = [
         },
     },
 ]
-
-
-def mcp_tools_to_openai_schemas(mcp_tools: list[dict]) -> list[dict]:
-    schemas: list[dict] = []
-    for tool in mcp_tools:
-        if not isinstance(tool, dict):
-            continue
-        name = tool.get("name")
-        if not isinstance(name, str) or not name:
-            continue
-        description = str(tool.get("description", "")).strip() or f"MCP tool: {name}"
-        input_schema = tool.get("inputSchema") or tool.get("input_schema")
-        if not isinstance(input_schema, dict):
-            input_schema = {"type": "object", "properties": {}, "required": []}
-        schemas.append(
-            {
-                "type": "function",
-                "function": {
-                    "name": name,
-                    "description": description,
-                    "parameters": input_schema,
-                },
-            }
-        )
-    return schemas

app/commands/builtins.py

@@ -325,7 +325,7 @@ async def echo(ctx: "CommandContext") -> None:
 @command_registry.command(
     "ai",
     "Ask a configurable AI model",
-    usage=".ai / ..ai [N] [--img] [--model] prompt",
+    usage=".ai / ..ai [N] [--model] prompt",
     category="ai",
 ).shared()
 async def ai(ctx: "CommandContext") -> None:
@@ -363,7 +363,6 @@ async def ai(ctx: "CommandContext") -> None:
             AIInvocation(
                 context_messages=invocation.context_messages,
                 model_flags=[selected_alias],
-                use_image_understanding=invocation.use_image_understanding,
                 prompt=invocation.prompt,
             ),
         )

app/config.py

@@ -52,14 +52,6 @@ class AIModelConfig:
     description: str | None = None
 
 
-@dataclass(slots=True)
-class AIImageUnderstandConfig:
-    api_key: str
-    api_host: str = "https://api.minimax.chat"
-    command: tuple[str, ...] = ("uvx", "minimax-coding-plan-mcp", "-y")
-    timeout: float | None = None
-
-
 @dataclass(slots=True)
 class AIPromptConfig:
     root: Path
@@ -73,7 +65,6 @@ class AIConfig:
     default_model: str
     context_messages: int
     prompt: AIPromptConfig
-    image_understand: AIImageUnderstandConfig | None = None
     providers: dict[str, AIProviderConfig] = field(default_factory=dict)
     models: dict[str, AIModelConfig] = field(default_factory=dict)
     workspace_dir: Path = field(default_factory=lambda: Path("workspace"))
@@ -136,22 +127,6 @@ def _parse_float(value: Any, *, field_name: str) -> float:
         raise ValueError(f"{field_name} must be a number") from exc
 
 
-def _parse_command_tokens(value: Any, *, field_name: str, default: tuple[str, ...]) -> tuple[str, ...]:
-    if value is None:
-        return default
-    if isinstance(value, str):
-        tokens = value.strip().split()
-        if not tokens:
-            raise ValueError(f"{field_name} must not be empty")
-        return tuple(tokens)
-    if isinstance(value, list):
-        tokens = [str(item).strip() for item in value if str(item).strip()]
-        if not tokens:
-            raise ValueError(f"{field_name} must not be empty")
-        return tuple(tokens)
-    raise ValueError(f"{field_name} must be a string or a list")
-
-
 def _parse_mapping(value: Any, *, field_name: str) -> dict[str, Any]:
     if value is None:
         return {}
@@ -252,31 +227,6 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
     if context_messages < 0:
         raise ValueError("ai.context_messages must be >= 0")
 
-    image_understand_raw = _parse_mapping(raw_ai.get("image_understand"), field_name="ai.image_understand")
-    image_understand: AIImageUnderstandConfig | None = None
-    if image_understand_raw:
-        api_key = str(image_understand_raw.get("api_key", "")).strip()
-        if not api_key:
-            raise ValueError("ai.image_understand.api_key is required")
-        api_host = str(image_understand_raw.get("api_host", "https://api.minimax.io")).strip()
-        if not api_host:
-            raise ValueError("ai.image_understand.api_host is required")
-        command = _parse_command_tokens(
-            image_understand_raw.get("command"),
-            field_name="ai.image_understand.command",
-            default=("uvx", "minimax-coding-plan-mcp", "-y"),
-        )
-        timeout_value = image_understand_raw.get("timeout")
-        timeout = None
-        if timeout_value is not None:
-            timeout = _parse_float(timeout_value, field_name="ai.image_understand.timeout")
-        image_understand = AIImageUnderstandConfig(
-            api_key=api_key,
-            api_host=api_host,
-            command=command,
-            timeout=timeout,
-        )
-
     providers_raw = _parse_mapping(raw_ai.get("providers"), field_name="ai.providers")
     models_raw = _parse_mapping(raw_ai.get("models"), field_name="ai.models")
 
@@ -341,7 +291,6 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
         default_model=default_model,
         context_messages=context_messages,
         prompt=prompt,
-        image_understand=image_understand,
         providers=providers,
         models=models,
         workspace_dir=workspace_dir,

config.example.json

@@ -23,11 +23,6 @@
       "parts_dir": "parts",
       "user_template": "[{full_name} · id:{user_id}]\n{text}"
     },
-    "image_understand": {
-      "api_key": "your_minimax_token_plan_key_here",
-      "api_host": "https://api.minimax.chat",
-      "command": ["uvx", "minimax-coding-plan-mcp", "-y"]
-    },
     "providers": {
       "minimax": {
         "api_key": "your_minimax_api_key_here",

prompts/ai/parts/25_images.md

@@ -1,5 +0,0 @@
-Если включён `--img`, используй результаты визуального анализа как обязательный контекст.
-
-- Не игнорируй найденные детали изображения, даже если текст истории уже кажется достаточным.
-- Если визуальный анализ противоречит тексту, укажи это явно.
-- Для ответа опирайся на все доступные изображения из выбранного окна контекста.