Convert AI responses to Telegram HTML with blockquote wrapping

Markdown (headings, bold, italic, strikethrough, code blocks, links)
is converted to Telegram-compatible HTML tags before sending. Each
reply chunk is wrapped in <blockquote> for visual distinction.

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

app/ai/service.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import asyncio
 from dataclasses import dataclass
 import inspect
 import json
 import logging
 from pathlib import Path
+import sys
 import tempfile
 from typing import TYPE_CHECKING, Any, Awaitable, Callable
 
@@ -19,7 +21,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.ai.tools import FILE_TOOL_DISPATCH, FILE_TOOL_SCHEMAS, PYTHON_EXEC_SCHEMA, mcp_tools_to_openai_schemas
 from app.config import AIConfig, AIModelConfig, AIProviderConfig
 
 if TYPE_CHECKING:
@@ -287,6 +289,27 @@ class AIService:
             ]
         ).strip()
 
+    def _render_user_message(self, text: str, user: Any) -> str:
+        template = self.config.prompt.user_template
+        if not template or not text:
+            return text
+        first_name = str(getattr(user, "first_name", "") or "")
+        last_name = str(getattr(user, "last_name", "") or "")
+        username = str(getattr(user, "username", "") or "")
+        user_id = str(getattr(user, "id", "") or "")
+        full_name = f"{first_name} {last_name}".strip() or username or user_id
+        try:
+            return template.format_map({
+                "text": text,
+                "user_id": user_id,
+                "first_name": first_name,
+                "last_name": last_name,
+                "username": username,
+                "full_name": full_name,
+            })
+        except (KeyError, ValueError):
+            return text
+
     async def build_messages(
         self,
         ctx: "CommandContext",
@@ -301,8 +324,13 @@ class AIService:
         for item in history:
             if not item.text:
                 continue
-            messages.append({"role": item.role, "content": item.text})
-        messages.append({"role": "user", "content": prompt})
+            if item.role == "user":
+                sender = getattr(item.message, "from_user", None)
+                content = self._render_user_message(item.text, sender)
+            else:
+                content = item.text
+            messages.append({"role": item.role, "content": content})
+        messages.append({"role": "user", "content": self._render_user_message(prompt, ctx.actor)})
         return messages
 
     async def _fetch_mcp_tool_schemas(self) -> list[dict]:
@@ -327,10 +355,93 @@ class AIService:
 
     async def _build_tool_list(self) -> list[dict]:
         mcp_schemas = await self._fetch_mcp_tool_schemas()
-        return FILE_TOOL_SCHEMAS + mcp_schemas
+        return FILE_TOOL_SCHEMAS + [PYTHON_EXEC_SCHEMA] + mcp_schemas
 
     def _get_allowed_roots(self) -> tuple[Path, ...]:
-        return (self.config.workspace_dir, self.config.prompt.root)
+        # [0] = base for resolution (project root); [1:] = permitted subdirs
+        project_root = self.config.workspace_dir.parent
+        return (project_root, self.config.workspace_dir, self.config.prompt.root)
+
+    def _build_exec_context(self, ctx: "CommandContext") -> dict[str, Any]:
+        actor = ctx.actor
+        replied = ctx.replied_message
+        replied_user = ctx.replied_user
+        return {
+            "chat_id": getattr(getattr(ctx.message, "chat", None), "id", None),
+            "message_id": getattr(ctx.message, "id", None),
+            "message_text": getattr(ctx.message, "text", None) or "",
+            "args": ctx.args,
+            "from_user": {
+                "id": getattr(actor, "id", None),
+                "first_name": getattr(actor, "first_name", None),
+                "last_name": getattr(actor, "last_name", None),
+                "username": getattr(actor, "username", None),
+            } if actor else None,
+            "replied_message": {
+                "id": getattr(replied, "id", None),
+                "text": getattr(replied, "text", None) or getattr(replied, "caption", None) or "",
+                "from_user": {
+                    "id": getattr(replied_user, "id", None),
+                    "first_name": getattr(replied_user, "first_name", None),
+                    "username": getattr(replied_user, "username", None),
+                } if replied_user else None,
+            } if replied else None,
+        }
+
+    async def _execute_python_exec(
+        self,
+        arguments: dict,
+        ctx: "CommandContext",
+    ) -> str:
+        code = str(arguments.get("code", "")).strip()
+        if not code:
+            raise ValueError("python_exec: 'code' is required")
+
+        raw_timeout = arguments.get("timeout", 10)
+        try:
+            timeout = float(raw_timeout)
+        except (TypeError, ValueError):
+            timeout = 10.0
+        timeout = min(max(timeout, 1.0), 30.0)
+
+        context = self._build_exec_context(ctx)
+        context_json = json.dumps(context, ensure_ascii=False, default=str)
+        context_line = f"import json as _json; ctx = _json.loads({context_json!r})\n"
+        full_code = (context_line + code).encode("utf-8")
+
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable, "-",
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            try:
+                stdout, stderr = await asyncio.wait_for(
+                    proc.communicate(input=full_code), timeout=timeout
+                )
+            except asyncio.TimeoutError:
+                proc.kill()
+                await proc.wait()
+                return f"Execution timed out after {timeout:.0f}s"
+        except OSError as exc:
+            raise ValueError(f"python_exec: failed to start interpreter: {exc}") from exc
+
+        out = stdout.decode("utf-8", errors="replace").rstrip()
+        err = stderr.decode("utf-8", errors="replace").rstrip()
+
+        parts: list[str] = []
+        if out:
+            parts.append(out)
+        if err:
+            parts.append(f"[stderr]\n{err}")
+        if not parts:
+            parts.append(f"[exit {proc.returncode}]" if proc.returncode else "(no output)")
+
+        result = "\n".join(parts)
+        if len(result) > 3000:
+            result = result[:3000] + "\n…(truncated)"
+        return result
 
     async def _execute_send_file(
         self,
@@ -373,7 +484,9 @@ class AIService:
             return f"Tool {name!r}: invalid JSON arguments: {exc}"
 
         try:
-            if name == "send_file":
+            if name == "python_exec":
+                result = await self._execute_python_exec(arguments, ctx)
+            elif name == "send_file":
                 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)

app/ai/tools.py

@@ -5,16 +5,23 @@ from typing import Any, Awaitable, Callable
 
 
 def _resolve_sandboxed_path(raw_path: str, allowed_roots: tuple[Path, ...]) -> Path:
+    """
+    allowed_roots[0]  — base directory used for resolving raw_path.
+    allowed_roots[1:] — permitted subdirectories; resolved path must fall inside one of them.
+    When only one root is given it acts as both base and permitted directory.
+    """
     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) + "/"):
+    base = allowed_roots[0]
+    allowed_dirs = allowed_roots[1:] if len(allowed_roots) > 1 else allowed_roots
+    candidate = (base / raw_path).resolve()
+    for allowed in allowed_dirs:
+        resolved = allowed.resolve()
+        if candidate == resolved or str(candidate).startswith(str(resolved) + "/"):
             return candidate
     raise ValueError(
         f"Path {raw_path!r} is outside the allowed directories "
-        f"({', '.join(str(r) for r in allowed_roots)})"
+        f"({', '.join(a.name for a in allowed_dirs)})"
     )
 
 
@@ -132,9 +139,10 @@ async def tool_list_files(arguments: dict, allowed_roots: tuple[Path, ...]) -> s
         if raw_path:
             base = _resolve_sandboxed_path(raw_path, allowed_roots)
         else:
-            # list all allowed roots combined
+            # list all allowed dirs (skip the base root at index 0)
+            dirs_to_list = allowed_roots[1:] if len(allowed_roots) > 1 else allowed_roots
             entries: list[str] = []
-            for root in allowed_roots:
+            for root in dirs_to_list:
                 if not root.exists():
                     continue
                 resolved_root = root.resolve()
@@ -184,6 +192,33 @@ def _walk(
         pass
 
 
+PYTHON_EXEC_SCHEMA: dict[str, Any] = {
+    "type": "function",
+    "function": {
+        "name": "python_exec",
+        "description": (
+            "Execute Python code and return stdout/stderr output. "
+            "A 'ctx' dict is pre-injected with message context: "
+            "chat_id, message_id, message_text, args, from_user, replied_message. "
+            "Use print() to produce output."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {
+                    "type": "string",
+                    "description": "Python code to execute",
+                },
+                "timeout": {
+                    "type": "number",
+                    "description": "Execution timeout in seconds (default 10, max 30)",
+                },
+            },
+            "required": ["code"],
+        },
+    },
+}
+
 ToolExecutor = Callable[[dict, tuple[Path, ...]], Awaitable[str]]
 
 FILE_TOOL_DISPATCH: dict[str, ToolExecutor] = {

app/commands/builtins.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import html
+import re
 from textwrap import wrap
 from typing import TYPE_CHECKING, Any
 
@@ -10,6 +12,40 @@ if TYPE_CHECKING:
     from app.commands.core import CommandContext
 
 
+_PH = "\x00{}\x00"
+_PH_RE = re.compile(r"\x00(\d+)\x00")
+
+
+def _md_to_tg(text: str) -> str:
+    """Convert markdown-ish text to Telegram HTML."""
+    saved: list[str] = []
+
+    def _save(s: str) -> str:
+        saved.append(s)
+        return _PH.format(len(saved) - 1)
+
+    def _fence(m: re.Match) -> str:
+        lang = (m.group(1) or "").strip()
+        code = html.escape(m.group(2) or "")
+        cls = f' class="language-{html.escape(lang)}"' if lang else ""
+        return _save(f"<pre><code{cls}>{code}</code></pre>")
+
+    text = re.sub(r"```(\w*)\n?(.*?)```", _fence, text, flags=re.DOTALL)
+    text = re.sub(r"`([^`\n]+)`", lambda m: _save(f"<code>{html.escape(m.group(1))}</code>"), text)
+
+    text = html.escape(text)
+
+    text = re.sub(r"^#{1,6}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE)
+    text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text, flags=re.DOTALL)
+    text = re.sub(r"__(.+?)__", r"<b>\1</b>", text, flags=re.DOTALL)
+    text = re.sub(r"(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)", r"<i>\1</i>", text, flags=re.DOTALL)
+    text = re.sub(r"(?<!_)_(?!_)(.+?)(?<!_)_(?!_)", r"<i>\1</i>", text, flags=re.DOTALL)
+    text = re.sub(r"~~(.+?)~~", r"<s>\1</s>", text, flags=re.DOTALL)
+    text = re.sub(r"\[(.+?)\]\((.+?)\)", r'<a href="\2">\1</a>', text)
+
+    return _PH_RE.sub(lambda m: saved[int(m.group(1))], text)
+
+
 command_registry = CommandRegistry()
 DEFAULT_REGISTRY = command_registry
 
@@ -69,11 +105,13 @@ async def _reply_long(ctx: "CommandContext", text: str) -> None:
     if not chunks:
         await ctx.reply("")
         return
-    first = await ctx.reply(chunks[0])
-    for chunk in chunks[1:]:
+    html_chunks = [f"<blockquote>{_md_to_tg(c)}</blockquote>" for c in chunks]
+    first = await ctx.reply(html_chunks[0], parse_mode="html")
+    for chunk in html_chunks[1:]:
         await ctx.client.send_message(
             ctx.message.chat.id,
             chunk,
+            parse_mode="html",
             reply_to_message_id=getattr(first, "id", None),
         )
 

app/config.py

@@ -65,6 +65,7 @@ class AIPromptConfig:
     root: Path
     system_file: str = "system.md"
     parts_dir: str = "parts"
+    user_template: str | None = None
 
 
 @dataclass(slots=True)
@@ -234,10 +235,13 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
     prompt_root = Path(str(prompt_raw.get("root", "prompts/ai"))).expanduser()
     if not prompt_root.is_absolute():
         prompt_root = (root / prompt_root).resolve()
+    raw_user_template = prompt_raw.get("user_template")
+    user_template = str(raw_user_template).strip() if raw_user_template else None
     prompt = AIPromptConfig(
         root=prompt_root,
         system_file=str(prompt_raw.get("system_file", "system.md")),
         parts_dir=str(prompt_raw.get("parts_dir", "parts")),
+        user_template=user_template or None,
     )
 
     default_model = str(raw_ai.get("default_model", "")).strip()

config.example.json

@@ -20,7 +20,8 @@
     "prompt": {
       "root": "prompts/ai",
       "system_file": "system.md",
-      "parts_dir": "parts"
+      "parts_dir": "parts",
+      "user_template": "[{full_name} · id:{user_id}]\n{text}"
     },
     "image_understand": {
       "api_key": "your_minimax_token_plan_key_here",