Add OpenAI tool-calling agentic loop with file tools and MiniMax MCP integration

- AIService.generate() now runs an agentic loop: sends tools to the API,
  handles tool_calls responses, executes tools, appends results, and repeats
  until a text response (up to tool_max_rounds, default 10)
- New app/ai/tools.py: six sandboxed file tools (write/read/create/modify/delete/list),
  OpenAI JSON schemas, path sandbox validator, and MCP→OpenAI schema converter
- Paths are sandboxed to workspace/ (project root) and prompts/ai/; traversal blocked
- MinimaxImageUnderstandClient gains list_tools() and call_tool() methods; MCP tool
  schemas are lazily fetched once per session and forwarded to the AI as callable tools
- AIConfig gains workspace_dir (Path, default workspace/) and tool_max_rounds (int, default 10)
- MCP client opened lazily only when an MCP tool is actually called during a loop

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -0,0 +1,56 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Running the bot
+
+```bash
+pip install -r requirements.txt
+python main.py
+```
+
+Logs go to `bot.log` by default (rotating, 10 MB, 3 backups). Override with `LOG_FILE=<path>`.
+
+## 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.
+
+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).
+
+## Architecture
+
+### Boot path
+`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.
+- **Registry** (`core.py`): `CommandRegistry` holds `CommandDefinition` objects. `DEFAULT_REGISTRY` is the singleton used at runtime; commands self-register via decorator.
+- **Built-in commands** (`builtins.py`): `.ping`, `.status`, `.help`, `.rights`/`.acl`, `.allow`, `.deny` (access), `.echo` (shared), `.ai` (shared). Add new commands here using the `@command_registry.command(...)` decorator pattern.
+- **Check system** (`core.py`): `CommandCheck` functions (e.g. `owner_only_check()`, `shared_access_check()`, `require_reply_check()`, `require_args_check()`) gate handler execution. Chain them with `.check(...)` or `.owner_only()` / `.require_reply()` / `.require_args()` on the builder.
+- **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.
+- **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.
+- **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`).
+- `AccountState` — runtime state per account (client, ACL store, status, `me` object).
+- `AIInvocation` — parsed AI call parameters (context_messages, model_flags, use_image_understanding, prompt).
+
+## Adding a new command
+
+```python
+@command_registry.command(
+    "mycommand",
+    "Short description",
+    usage=".mycommand <arg>",
+    category="general",
+).shared()          # omit for owner-only; add .require_args() / .require_reply() as needed
+async def mycommand(ctx: "CommandContext") -> None:
+    await ctx.reply(ctx.args)
+```
+
+Shared commands use `..mycommand` for guests and `.mycommand` for the owner. Owner-only commands use `.mycommand` only.

app/ai/image_understand.py

@@ -187,6 +187,31 @@ class MinimaxImageUnderstandClient:
         self._stdin = None
         self._stdout = None
 
+    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()

app/ai/service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 import inspect
+import json
 import logging
 from pathlib import Path
 import tempfile
@@ -18,6 +19,7 @@ from app.ai.image_understand import (
 )
 from app.ai.parser import AIInvocation
 from app.ai.prompts import PromptBook
+from app.ai.tools import FILE_TOOL_DISPATCH, FILE_TOOL_SCHEMAS, mcp_tools_to_openai_schemas
 from app.config import AIConfig, AIModelConfig, AIProviderConfig
 
 if TYPE_CHECKING:
@@ -66,6 +68,8 @@ 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
 
     @property
     def enabled(self) -> bool:
@@ -287,8 +291,8 @@ class AIService:
         prompt: str,
         history: list[HistoryMessage],
         image_notes: list[str],
-    ) -> list[dict[str, str]]:
-        messages: list[dict[str, str]] = [{"role": "system", "content": self.load_system_prompt()}]
+    ) -> 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:
@@ -298,6 +302,60 @@ class AIService:
         messages.append({"role": "user", "content": prompt})
         return messages
 
+    async def _fetch_mcp_tool_schemas(self) -> list[dict]:
+        if self._mcp_tools_fetched:
+            return self._mcp_tool_schemas or []
+        self._mcp_tools_fetched = True
+        if self.config.image_understand is None:
+            self._mcp_tool_schemas = []
+            return []
+        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)
+        except Exception:
+            log.warning("Failed to fetch MCP tool schemas; MiniMax tools will be unavailable")
+            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 + mcp_schemas
+
+    def _get_allowed_roots(self) -> tuple[Path, ...]:
+        return (self.config.workspace_dir, self.config.prompt.root)
+
+    async def _execute_tool_call(
+        self,
+        tool_call: Any,
+        mcp_client: MinimaxImageUnderstandClient | None,
+        mcp_names: set[str],
+        allowed_roots: tuple[Path, ...],
+        progress: ProgressCallback | None,
+    ) -> str:
+        name = getattr(getattr(tool_call, "function", None), "name", None) or ""
+        raw_args = getattr(getattr(tool_call, "function", None), "arguments", None) or "{}"
+
+        try:
+            arguments: dict = json.loads(raw_args)
+        except json.JSONDecodeError as exc:
+            return f"Tool {name!r}: invalid JSON arguments: {exc}"
+
+        try:
+            if 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
+
     async def generate(
         self,
         ctx: "CommandContext",
@@ -327,20 +385,90 @@ class AIService:
             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
+
         try:
-            response = await client.chat.completions.create(
-                model=model.model.model,
-                messages=messages,
-            )
-        except Exception as exc:
-            log.exception("AI request failed for model %s", model.alias)
-            raise RuntimeError(f"AI request failed: {exc}") from exc
+            for round_num in range(max_rounds + 1):
+                kwargs: dict[str, Any] = {
+                    "model": model.model.model,
+                    "messages": messages,
+                }
+                if tools:
+                    kwargs["tools"] = tools
 
-        text = self._extract_response_text(response)
-        if not text:
-            raise RuntimeError("AI model returned an empty response")
+                try:
+                    response = await client.chat.completions.create(**kwargs)
+                except Exception as exc:
+                    log.exception("AI request failed for model %s", model.alias)
+                    raise RuntimeError(f"AI request failed: {exc}") from exc
 
-        return AIResponse(text=text, model=model.model, provider=model.provider)
+                choice = response.choices[0]
+                finish_reason = getattr(choice, "finish_reason", None)
+
+                if finish_reason != "tool_calls":
+                    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 round_num >= max_rounds:
+                    raise RuntimeError(
+                        f"Agentic loop exceeded the maximum of {max_rounds} tool rounds"
+                    )
+
+                tool_calls = getattr(choice.message, "tool_calls", None) or []
+                if not tool_calls:
+                    raise RuntimeError("finish_reason was tool_calls but no tool_calls present")
+
+                messages.append({
+                    "role": "assistant",
+                    "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
+                    ],
+                })
+
+                needs_mcp = any(
+                    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, 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

@@ -0,0 +1,347 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+
+def _resolve_sandboxed_path(raw_path: str, allowed_roots: tuple[Path, ...]) -> Path:
+    if raw_path.startswith("/"):
+        raise ValueError(f"Absolute paths are not allowed: {raw_path!r}")
+    for root in allowed_roots:
+        resolved_root = root.resolve()
+        candidate = (root / raw_path).resolve()
+        if candidate == resolved_root or str(candidate).startswith(str(resolved_root) + "/"):
+            return candidate
+    raise ValueError(
+        f"Path {raw_path!r} is outside the allowed directories "
+        f"({', '.join(str(r) for r in allowed_roots)})"
+    )
+
+
+async def tool_write_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    content = str(arguments.get("content", ""))
+    if not raw_path:
+        raise ValueError("write_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        return f"Written {len(content.encode('utf-8'))} bytes to {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"write_file failed: {exc.strerror}") from exc
+
+
+async def tool_read_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    if not raw_path:
+        raise ValueError("read_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if not path.exists():
+            raise ValueError(f"read_file: file not found: {raw_path!r}")
+        if not path.is_file():
+            raise ValueError(f"read_file: not a file: {raw_path!r}")
+        size = path.stat().st_size
+        if size > 100 * 1024:
+            raise ValueError(f"read_file: file exceeds the 100 KB read limit ({size} bytes)")
+        try:
+            return path.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            return "File is binary (not readable as text)"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"read_file failed: {exc.strerror}") from exc
+
+
+async def tool_create_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    content = str(arguments.get("content", ""))
+    if not raw_path:
+        raise ValueError("create_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if path.exists():
+            raise ValueError(f"create_file: file already exists: {raw_path!r}")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        return f"Created {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"create_file failed: {exc.strerror}") from exc
+
+
+async def tool_modify_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    operation = str(arguments.get("operation", "")).strip().lower()
+    if not raw_path:
+        raise ValueError("modify_file: 'path' is required")
+    if operation not in ("append", "replace"):
+        raise ValueError("modify_file: 'operation' must be 'append' or 'replace'")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if operation == "append":
+            text = str(arguments.get("text", ""))
+            existing = path.read_text(encoding="utf-8") if path.exists() else ""
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(existing + text, encoding="utf-8")
+            return f"Appended {len(text.encode('utf-8'))} bytes to {raw_path}"
+        else:
+            old_text = arguments.get("old_text")
+            new_text = arguments.get("new_text")
+            if old_text is None or new_text is None:
+                raise ValueError("modify_file replace: 'old_text' and 'new_text' are required")
+            old_text = str(old_text)
+            new_text = str(new_text)
+            if not path.exists():
+                raise ValueError(f"modify_file: file not found: {raw_path!r}")
+            content = path.read_text(encoding="utf-8")
+            if old_text not in content:
+                raise ValueError(f"modify_file replace: 'old_text' not found in {raw_path!r}")
+            path.write_text(content.replace(old_text, new_text, 1), encoding="utf-8")
+            return f"Replaced text in {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"modify_file failed: {exc.strerror}") from exc
+
+
+async def tool_delete_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    if not raw_path:
+        raise ValueError("delete_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if not path.exists():
+            raise ValueError(f"delete_file: file not found: {raw_path!r}")
+        path.unlink()
+        return f"Deleted {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"delete_file failed: {exc.strerror}") from exc
+
+
+async def tool_list_files(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    try:
+        if raw_path:
+            base = _resolve_sandboxed_path(raw_path, allowed_roots)
+        else:
+            # list all allowed roots combined
+            entries: list[str] = []
+            for root in allowed_roots:
+                if not root.exists():
+                    continue
+                resolved_root = root.resolve()
+                entries.extend(_collect_files(root, resolved_root, max_depth=3))
+            return "\n".join(sorted(entries)) or "(empty)"
+
+        if not base.exists():
+            return "(empty)"
+        if not base.is_dir():
+            raise ValueError(f"list_files: not a directory: {raw_path!r}")
+        resolved_base = base.resolve()
+        entries = _collect_files(base, resolved_base, max_depth=3)
+        return "\n".join(sorted(entries)) or "(empty)"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"list_files failed: {exc.strerror}") from exc
+
+
+def _collect_files(directory: Path, resolved_root: Path, *, max_depth: int) -> list[str]:
+    result: list[str] = []
+    _walk(directory, resolved_root, directory, max_depth, 0, result)
+    return result
+
+
+def _walk(
+    current: Path,
+    resolved_root: Path,
+    base: Path,
+    max_depth: int,
+    depth: int,
+    result: list[str],
+) -> None:
+    if depth > max_depth:
+        return
+    try:
+        for entry in current.iterdir():
+            resolved_entry = entry.resolve()
+            # skip symlinks that escape the root
+            if not str(resolved_entry).startswith(str(resolved_root)):
+                continue
+            if entry.is_file():
+                result.append(str(entry.relative_to(base)))
+            elif entry.is_dir():
+                _walk(entry, resolved_root, base, max_depth, depth + 1, result)
+    except OSError:
+        pass
+
+
+ToolExecutor = Callable[[dict, tuple[Path, ...]], Awaitable[str]]
+
+FILE_TOOL_DISPATCH: dict[str, ToolExecutor] = {
+    "write_file": tool_write_file,
+    "read_file": tool_read_file,
+    "create_file": tool_create_file,
+    "modify_file": tool_modify_file,
+    "delete_file": tool_delete_file,
+    "list_files": tool_list_files,
+}
+
+FILE_TOOL_SCHEMAS: list[dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": (
+                "Create or overwrite a file. Allowed path prefixes: workspace/ or prompts/. "
+                "Example path: workspace/notes.txt or prompts/ai/parts/extra.md"
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "content": {"type": "string", "description": "File content to write"},
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "read_file",
+            "description": "Read the content of a file (max 100 KB). Allowed path prefixes: workspace/ or prompts/.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "create_file",
+            "description": (
+                "Create a new file. Fails if the file already exists. "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "content": {"type": "string", "description": "Initial file content"},
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "modify_file",
+            "description": (
+                "Modify an existing file. Two operations: "
+                "'append' adds text at end (requires 'text'); "
+                "'replace' substitutes first occurrence of old_text with new_text "
+                "(requires 'old_text' and 'new_text', fails if old_text not found). "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "operation": {
+                        "type": "string",
+                        "enum": ["append", "replace"],
+                        "description": "Operation to perform",
+                    },
+                    "text": {
+                        "type": "string",
+                        "description": "Text to append (used when operation=append)",
+                    },
+                    "old_text": {
+                        "type": "string",
+                        "description": "Text to find (used when operation=replace)",
+                    },
+                    "new_text": {
+                        "type": "string",
+                        "description": "Replacement text (used when operation=replace)",
+                    },
+                },
+                "required": ["path", "operation"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "delete_file",
+            "description": "Delete a file. Allowed path prefixes: workspace/ or prompts/.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "list_files",
+            "description": (
+                "List files recursively (up to 3 levels deep). "
+                "Pass an empty path or omit to list all allowed directories. "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {
+                        "type": "string",
+                        "description": "Directory path to list (relative). Leave empty to list all.",
+                        "default": "",
+                    },
+                },
+                "required": [],
+            },
+        },
+    },
+]
+
+
+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/config.py

@@ -71,6 +71,8 @@ class AIConfig:
     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"))
+    tool_max_rounds: int = 10
 
 
 @dataclass(slots=True)
@@ -289,6 +291,15 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
                 f"ai.models.{alias}.provider references unknown provider: {model_config.provider}"
             )
 
+    workspace_dir_raw = raw_ai.get("workspace_dir", "workspace")
+    workspace_dir = Path(str(workspace_dir_raw)).expanduser()
+    if not workspace_dir.is_absolute():
+        workspace_dir = (root / workspace_dir).resolve()
+
+    tool_max_rounds = _parse_int(raw_ai.get("tool_max_rounds", 10), field_name="ai.tool_max_rounds")
+    if tool_max_rounds < 1:
+        raise ValueError("ai.tool_max_rounds must be >= 1")
+
     return AIConfig(
         default_model=default_model,
         context_messages=context_messages,
@@ -296,6 +307,8 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
         image_understand=image_understand,
         providers=providers,
         models=models,
+        workspace_dir=workspace_dir,
+        tool_max_rounds=tool_max_rounds,
     )
 
 

config.example.json

@@ -15,6 +15,8 @@
   "ai": {
     "default_model": "mm",
     "context_messages": 12,
+    "workspace_dir": "workspace",
+    "tool_max_rounds": 10,
     "prompt": {
       "root": "prompts/ai",
       "system_file": "system.md",