feat(agent): unify tool calling around MCP and add /api/mcp endpoint

domain/agent/__init__.py

@@ -1,9 +1,10 @@
 from .service import AgentService
-from .types import AgentChatContext, AgentChatRequest, PendingToolCall
+from .types import AgentChatContext, AgentChatRequest, McpCall, PendingMcpCall
 
 __all__ = [
     "AgentService",
     "AgentChatContext",
     "AgentChatRequest",
-    "PendingToolCall",
+    "McpCall",
+    "PendingMcpCall",
 ]

domain/agent/api.py

@@ -14,7 +14,7 @@ router = APIRouter(prefix="/api/agent", tags=["agent"])
 
 
 @router.post("/chat")
-@audit(action=AuditAction.CREATE, description="Agent 对话", body_fields=["auto_execute"])
+@audit(action=AuditAction.CREATE, description="Agent 对话", body_fields=["auto_execute", "approved_mcp_call_ids", "rejected_mcp_call_ids"])
 async def chat(
     request: Request,
     payload: AgentChatRequest,
@@ -25,7 +25,7 @@ async def chat(
 
 
 @router.post("/chat/stream")
-@audit(action=AuditAction.CREATE, description="Agent 对话（SSE）", body_fields=["auto_execute"])
+@audit(action=AuditAction.CREATE, description="Agent 对话（SSE）", body_fields=["auto_execute", "approved_mcp_call_ids", "rejected_mcp_call_ids"])
 async def chat_stream(
     request: Request,
     payload: AgentChatRequest,

domain/agent/mcp.py

@@ -0,0 +1,334 @@
+import inspect
+import json
+from contextlib import asynccontextmanager
+from datetime import timedelta
+from typing import Annotated, Any, Literal
+from urllib.parse import quote, unquote
+
+import httpx
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+from mcp.server.auth.provider import AccessToken
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.fastmcp.server import AuthSettings
+from mcp.types import ToolAnnotations
+from pydantic import Field
+
+from domain.auth import AuthService, User
+from domain.processors import ProcessorService
+
+from .tools import get_tool, mcp_tool_descriptors
+from .tools.base import McpToolDescriptor, normalize_tool_result, tool_result_to_content
+
+INTERNAL_MCP_BASE_URL = "http://127.0.0.1:8000/"
+CURRENT_PATH_HEADER = "x-foxel-current-path"
+
+
+def _normalize_path(path: str | None) -> str | None:
+    if not path:
+        return None
+    value = str(path).strip().replace("\\", "/")
+    if not value:
+        return None
+    if not value.startswith("/"):
+        value = "/" + value
+    return value.rstrip("/") or "/"
+
+
+def _header_current_path(ctx: Context | None) -> str | None:
+    request = ctx.request_context.request if ctx and ctx.request_context else None
+    if request is None:
+        return None
+    return _normalize_path(request.headers.get(CURRENT_PATH_HEADER))
+
+
+def _field_annotation(schema: dict[str, Any], required: bool) -> tuple[Any, Any]:
+    raw_type = schema.get("type")
+    enum_values = schema.get("enum")
+    description = str(schema.get("description") or "").strip() or None
+    default = schema.get("default", inspect.Parameter.empty if required else None)
+
+    annotation: Any
+    if isinstance(enum_values, list) and enum_values:
+        annotation = Literal.__getitem__(tuple(enum_values))
+    elif raw_type == "string":
+        annotation = str
+    elif raw_type == "integer":
+        annotation = int
+    elif raw_type == "number":
+        annotation = float
+    elif raw_type == "boolean":
+        annotation = bool
+    elif raw_type == "array":
+        annotation = list[Any]
+    elif raw_type == "object":
+        annotation = dict[str, Any]
+    else:
+        annotation = Any
+
+    if not required and default is None:
+        annotation = annotation | None
+
+    if description:
+        annotation = Annotated[annotation, Field(description=description)]
+    return annotation, default
+
+
+def _build_tool_signature(descriptor: McpToolDescriptor) -> inspect.Signature:
+    schema = descriptor.input_schema if isinstance(descriptor.input_schema, dict) else {}
+    properties = schema.get("properties") if isinstance(schema.get("properties"), dict) else {}
+    required = set(schema.get("required") or [])
+    parameters: list[inspect.Parameter] = []
+    for key, value in properties.items():
+        prop_schema = value if isinstance(value, dict) else {}
+        annotation, default = _field_annotation(prop_schema, key in required)
+        parameters.append(
+            inspect.Parameter(
+                str(key),
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                default=default,
+                annotation=annotation,
+            )
+        )
+    return inspect.Signature(parameters=parameters, return_annotation=dict[str, Any])
+
+
+def _build_tool_wrapper(descriptor: McpToolDescriptor):
+    async def wrapper(**kwargs: Any) -> dict[str, Any]:
+        spec = get_tool(descriptor.name)
+        if not spec:
+            return normalize_tool_result({"error": f"unknown_tool: {descriptor.name}"})
+        try:
+            result = await spec.handler(kwargs)
+            return normalize_tool_result(result)
+        except Exception as exc:  # noqa: BLE001
+            return normalize_tool_result({"error": str(exc)})
+
+    wrapper.__name__ = descriptor.name
+    wrapper.__doc__ = descriptor.description
+    wrapper.__signature__ = _build_tool_signature(descriptor)
+    return wrapper
+
+
+class FoxelMcpTokenVerifier:
+    async def verify_token(self, token: str) -> AccessToken | None:
+        try:
+            user = await AuthService.get_current_active_user(await AuthService.get_current_user(token))
+        except Exception:  # noqa: BLE001
+            return None
+        return AccessToken(token=token, client_id=user.username, scopes=[])
+
+
+MCP_SERVER = FastMCP(
+    name="Foxel MCP",
+    instructions="Foxel 内置 MCP 服务，提供文件系统、网页抓取、时间与处理器相关能力。",
+    streamable_http_path="/",
+    token_verifier=FoxelMcpTokenVerifier(),
+    auth=AuthSettings(
+        issuer_url="http://127.0.0.1:8000",
+        resource_server_url=None,
+        required_scopes=[],
+    ),
+)
+
+
+for descriptor in mcp_tool_descriptors():
+    MCP_SERVER.add_tool(
+        _build_tool_wrapper(descriptor),
+        name=descriptor.name,
+        description=descriptor.description,
+        annotations=ToolAnnotations.model_validate(descriptor.annotations),
+        meta=descriptor.meta,
+        structured_output=False,
+    )
+
+
+@MCP_SERVER.resource(
+    "foxel://context/current-path",
+    name="current_path",
+    title="Current Path",
+    description="返回当前请求上下文里的文件管理目录。",
+    mime_type="application/json",
+)
+def current_path_resource() -> dict[str, Any]:
+    return {"current_path": None}
+
+
+@MCP_SERVER.resource(
+    "foxel://policy/tool-confirmation",
+    name="tool_confirmation_policy",
+    title="Tool Confirmation Policy",
+    description="返回 Foxel agent 对工具审批的策略。",
+    mime_type="application/json",
+)
+def tool_confirmation_policy_resource() -> dict[str, Any]:
+    return {
+        "read_tools": [tool.name for tool in mcp_tool_descriptors() if not tool.requires_confirmation],
+        "write_tools": [tool.name for tool in mcp_tool_descriptors() if tool.requires_confirmation],
+        "rule": "直接调用 MCP tool 时不额外审批；通过 agent 代表用户执行写操作时需要审批。",
+    }
+
+
+@MCP_SERVER.resource(
+    "foxel://processors/index",
+    name="processors_index",
+    title="Processors Index",
+    description="返回当前可用处理器列表。",
+    mime_type="application/json",
+)
+def processors_index_resource() -> dict[str, Any]:
+    return {"processors": ProcessorService.list_processors()}
+
+
+async def _tool_resource(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+    spec = get_tool(tool_name)
+    if not spec:
+        return normalize_tool_result({"error": f"unknown_tool: {tool_name}"})
+    try:
+        result = await spec.handler(arguments)
+        return normalize_tool_result(result)
+    except Exception as exc:  # noqa: BLE001
+        return normalize_tool_result({"error": str(exc)})
+
+
+@MCP_SERVER.resource(
+    "foxel://vfs/stat/{path}",
+    name="vfs_stat_resource",
+    title="VFS Stat",
+    description="读取指定路径的文件或目录元信息；path 需要 URL 编码。",
+    mime_type="application/json",
+)
+async def vfs_stat_resource(path: str) -> dict[str, Any]:
+    return await _tool_resource("vfs_stat", {"path": "/" + unquote(path).lstrip("/")})
+
+
+@MCP_SERVER.resource(
+    "foxel://vfs/text/{path}",
+    name="vfs_text_resource",
+    title="VFS Text",
+    description="读取文本文件内容；path 需要 URL 编码。",
+    mime_type="application/json",
+)
+async def vfs_text_resource(path: str) -> dict[str, Any]:
+    return await _tool_resource("vfs_read_text", {"path": "/" + unquote(path).lstrip("/")})
+
+
+@MCP_SERVER.resource(
+    "foxel://vfs/dir/{path}",
+    name="vfs_dir_resource",
+    title="VFS Directory",
+    description="列出目录内容；path 需要 URL 编码。",
+    mime_type="application/json",
+)
+async def vfs_dir_resource(path: str) -> dict[str, Any]:
+    return await _tool_resource("vfs_list_dir", {"path": "/" + unquote(path).lstrip("/")})
+
+
+@MCP_SERVER.resource(
+    "foxel://vfs/search/{query}",
+    name="vfs_search_resource",
+    title="VFS Search",
+    description="搜索文件；query 需要 URL 编码。",
+    mime_type="application/json",
+)
+async def vfs_search_resource(query: str) -> dict[str, Any]:
+    return await _tool_resource("vfs_search", {"q": unquote(query)})
+
+
+@MCP_SERVER.prompt(name="browse_path", title="Browse Path", description="生成浏览目录的推荐提示词。")
+def browse_path_prompt(path: Annotated[str, Field(description="目标目录路径")]) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请先浏览目录 `{path}`，总结结构与关键文件。必要时调用 vfs_list_dir 与 vfs_stat。"}]
+
+
+@MCP_SERVER.prompt(name="inspect_file", title="Inspect File", description="生成查看文件的推荐提示词。")
+def inspect_file_prompt(path: Annotated[str, Field(description="目标文件路径")]) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请检查文件 `{path}` 的内容与用途。必要时调用 vfs_read_text。"}]
+
+
+@MCP_SERVER.prompt(name="search_files", title="Search Files", description="生成搜索文件的推荐提示词。")
+def search_files_prompt(query: Annotated[str, Field(description="搜索关键词")]) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请搜索与 `{query}` 相关的文件，并按相关性总结。必要时调用 vfs_search。"}]
+
+
+@MCP_SERVER.prompt(name="edit_file_safely", title="Edit File Safely", description="生成安全修改文件的推荐提示词。")
+def edit_file_safely_prompt(path: Annotated[str, Field(description="目标文件路径")]) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请先读取 `{path}`，解释拟修改点，再等待我确认后执行写入。"}]
+
+
+@MCP_SERVER.prompt(name="run_processor", title="Run Processor", description="生成运行处理器的推荐提示词。")
+def run_processor_prompt(
+    path: Annotated[str, Field(description="目标文件或目录路径")],
+    processor_type: Annotated[str, Field(description="处理器类型")],
+) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请检查 `{path}` 是否适合运行处理器 `{processor_type}`，确认参数后再执行 processors_run。"}]
+
+
+@MCP_SERVER.prompt(name="fetch_web_page", title="Fetch Web Page", description="生成抓取网页的推荐提示词。")
+def fetch_web_page_prompt(url: Annotated[str, Field(description="目标网址")]) -> list[dict[str, Any]]:
+    return [{"role": "user", "content": f"请抓取网页 `{url}`，并总结标题、正文与关键链接。必要时调用 web_fetch。"}]
+
+
+MCP_HTTP_APP = MCP_SERVER.streamable_http_app()
+
+
+def loopback_httpx_client_factory(app):
+    def factory(headers: dict[str, str] | None = None, timeout=None, auth=None) -> httpx.AsyncClient:
+        return httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app),
+            base_url=INTERNAL_MCP_BASE_URL.rstrip("/"),
+            headers=headers,
+            timeout=timeout,
+            auth=auth,
+            follow_redirects=True,
+        )
+
+    return factory
+
+
+async def create_loopback_mcp_headers(user: User | None, current_path: str | None = None) -> dict[str, str]:
+    headers: dict[str, str] = {}
+    if user is not None:
+        token = await AuthService.create_access_token(
+            {"sub": user.username},
+            expires_delta=timedelta(minutes=5),
+        )
+        headers["Authorization"] = f"Bearer {token}"
+    if current_path:
+        headers[CURRENT_PATH_HEADER] = current_path
+    return headers
+
+
+@asynccontextmanager
+async def mcp_client_session(user: User | None, current_path: str | None = None):
+    headers = await create_loopback_mcp_headers(user, current_path)
+    async with streamablehttp_client(
+        INTERNAL_MCP_BASE_URL,
+        headers=headers,
+        httpx_client_factory=loopback_httpx_client_factory(MCP_HTTP_APP),
+    ) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            yield session
+
+
+def mcp_content_to_text(content: list[Any], structured_content: dict[str, Any] | None = None) -> str:
+    if structured_content is not None:
+        try:
+            return json.dumps(structured_content, ensure_ascii=False)
+        except TypeError:
+            pass
+
+    text_parts: list[str] = []
+    for item in content:
+        item_type = getattr(item, "type", None)
+        if item_type == "text":
+            text = getattr(item, "text", None)
+            if isinstance(text, str) and text:
+                text_parts.append(text)
+    if text_parts:
+        return "\n".join(text_parts)
+    return tool_result_to_content({"error": "empty_mcp_content"})
+
+
+def encode_resource_path(path: str) -> str:
+    return quote(path.lstrip("/"), safe="")

domain/agent/service.py

@@ -8,27 +8,27 @@ from fastapi import HTTPException
 
 from domain.ai import AIProviderService, MissingModelError, chat_completion, chat_completion_stream
 from domain.auth import User
-from .tools import get_tool, openai_tools, tool_result_to_content
-from .types import AgentChatRequest, PendingToolCall
+
+from .mcp import mcp_client_session, mcp_content_to_text
+from .tools import tool_result_to_content
+from .types import AgentChatRequest, PendingMcpCall
 
 
-def _normalize_path(p: Optional[str]) -> Optional[str]:
-    if not p:
+def _normalize_path(path: Optional[str]) -> Optional[str]:
+    if not path:
         return None
-    s = str(p).strip()
-    if not s:
+    value = str(path).strip().replace("\\", "/")
+    if not value:
         return None
-    s = s.replace("\\", "/")
-    if not s.startswith("/"):
-        s = "/" + s
-    s = s.rstrip("/") or "/"
-    return s
+    if not value.startswith("/"):
+        value = "/" + value
+    return value.rstrip("/") or "/"
 
 
 def _build_system_prompt(current_path: Optional[str]) -> str:
     lines = [
         "你是 Foxel 的 AI 助手。",
-        "你可以通过工具对文件/目录进行查询、读写、移动、复制、删除，以及运行处理器（processor）。",
+        "你可以通过 MCP 工具对文件/目录进行查询、读写、移动、复制、删除，以及运行处理器（processor）。",
         "",
         "可用工具：",
         "- time：获取服务器当前时间（精确到秒，英文星期），支持 year/month/day/hour/minute/second 偏移。",
@@ -60,13 +60,13 @@ def _build_system_prompt(current_path: Optional[str]) -> str:
     return "\n".join(lines)
 
 
-def _ensure_tool_call_ids(message: Dict[str, Any]) -> Dict[str, Any]:
-    tool_calls = message.get("tool_calls")
-    if not isinstance(tool_calls, list):
+def _ensure_mcp_call_ids(message: Dict[str, Any]) -> Dict[str, Any]:
+    mcp_calls = message.get("mcp_calls")
+    if not isinstance(mcp_calls, list):
         return message
 
     changed = False
-    for idx, call in enumerate(tool_calls):
+    for idx, call in enumerate(mcp_calls):
         if not isinstance(call, dict):
             continue
         call_id = call.get("id")
@@ -76,57 +76,54 @@ def _ensure_tool_call_ids(message: Dict[str, Any]) -> Dict[str, Any]:
         changed = True
 
     if changed:
-        message["tool_calls"] = tool_calls
+        message["mcp_calls"] = mcp_calls
     return message
 
 
-def _extract_pending(tool_call: Dict[str, Any], requires_confirmation: bool) -> PendingToolCall:
-    call_id = str(tool_call.get("id") or "")
-    fn = tool_call.get("function") or {}
-    name = str((fn.get("name") if isinstance(fn, dict) else None) or "")
-    raw_args = fn.get("arguments") if isinstance(fn, dict) else None
-    arguments: Dict[str, Any] = {}
-    if isinstance(raw_args, str) and raw_args.strip():
-        try:
-            parsed = json.loads(raw_args)
-            if isinstance(parsed, dict):
-                arguments = parsed
-        except json.JSONDecodeError:
-            arguments = {}
-    return PendingToolCall(
-        id=call_id,
-        name=name,
+def _extract_pending(mcp_call: Dict[str, Any], requires_confirmation: bool) -> PendingMcpCall:
+    arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {}
+    return PendingMcpCall(
+        id=str(mcp_call.get("id") or ""),
+        name=str(mcp_call.get("name") or ""),
         arguments=arguments,
         requires_confirmation=requires_confirmation,
     )
 
 
-def _find_last_assistant_tool_calls(messages: List[Dict[str, Any]]) -> Tuple[int, Dict[str, Any]]:
+def _find_last_assistant_mcp_calls(messages: List[Dict[str, Any]]) -> Tuple[int, Dict[str, Any]]:
     for idx in range(len(messages) - 1, -1, -1):
         msg = messages[idx]
         if not isinstance(msg, dict):
             continue
         if msg.get("role") != "assistant":
             continue
-        tool_calls = msg.get("tool_calls")
-        if isinstance(tool_calls, list) and tool_calls:
+        mcp_calls = msg.get("mcp_calls")
+        if isinstance(mcp_calls, list) and mcp_calls:
             return idx, msg
     raise HTTPException(status_code=400, detail="没有可确认的待执行操作")
 
 
-def _existing_tool_result_ids(messages: List[Dict[str, Any]]) -> set[str]:
+def _existing_mcp_result_ids(messages: List[Dict[str, Any]]) -> set[str]:
     ids: set[str] = set()
     for msg in messages:
         if not isinstance(msg, dict):
             continue
         if msg.get("role") != "tool":
             continue
-        tool_call_id = msg.get("tool_call_id")
-        if isinstance(tool_call_id, str) and tool_call_id.strip():
-            ids.add(tool_call_id)
+        call_id = msg.get("mcp_call_id")
+        if isinstance(call_id, str) and call_id.strip():
+            ids.add(call_id)
     return ids
 
 
+def _tool_requires_confirmation(tool_descriptor: Dict[str, Any]) -> bool:
+    meta = tool_descriptor.get("meta") if isinstance(tool_descriptor.get("meta"), dict) else {}
+    if "requires_confirmation" in meta:
+        return bool(meta.get("requires_confirmation"))
+    annotations = tool_descriptor.get("annotations") if isinstance(tool_descriptor.get("annotations"), dict) else {}
+    return not bool(annotations.get("readOnlyHint"))
+
+
 async def _choose_chat_ability() -> str:
     tools_model = await AIProviderService.get_default_model("tools")
     return "tools" if tools_model else "chat"
@@ -142,245 +139,91 @@ def _format_exc(exc: BaseException) -> str:
     return text if text else exc.__class__.__name__
 
 
+async def _list_mcp_tools(session) -> List[Dict[str, Any]]:
+    result = await session.list_tools()
+    tools: List[Dict[str, Any]] = []
+    for item in result.tools:
+        annotations = getattr(item, "annotations", None)
+        meta = getattr(item, "meta", None)
+        tools.append(
+            {
+                "name": str(getattr(item, "name", "") or ""),
+                "description": str(getattr(item, "description", "") or ""),
+                "input_schema": getattr(item, "inputSchema", None) or {},
+                "annotations": annotations.model_dump(exclude_none=True) if annotations is not None else {},
+                "meta": meta if isinstance(meta, dict) else {},
+            }
+        )
+    return tools
+
+
+async def _execute_mcp_call(session, name: str, arguments: Dict[str, Any]) -> str:
+    result = await session.call_tool(name, arguments)
+    return mcp_content_to_text(result.content, result.structuredContent)
+
+
 class AgentService:
     @classmethod
     async def chat(cls, req: AgentChatRequest, user: Optional[User]) -> Dict[str, Any]:
         history: List[Dict[str, Any]] = list(req.messages or [])
         current_path = _normalize_path(req.context.current_path if req.context else None)
-
         system_prompt = _build_system_prompt(current_path)
         internal_messages: List[Dict[str, Any]] = [{"role": "system", "content": system_prompt}] + history
-
         new_messages: List[Dict[str, Any]] = []
-        pending: List[PendingToolCall] = []
+        pending: List[PendingMcpCall] = []
 
-        approved_ids = {i for i in (req.approved_tool_call_ids or []) if isinstance(i, str) and i.strip()}
-        rejected_ids = {i for i in (req.rejected_tool_call_ids or []) if isinstance(i, str) and i.strip()}
+        approved_ids = {i for i in (req.approved_mcp_call_ids or []) if isinstance(i, str) and i.strip()}
+        rejected_ids = {i for i in (req.rejected_mcp_call_ids or []) if isinstance(i, str) and i.strip()}
 
-        if approved_ids or rejected_ids:
-            _, last_call_msg = _find_last_assistant_tool_calls(internal_messages)
-            last_call_msg = _ensure_tool_call_ids(last_call_msg)
-            tool_calls = last_call_msg.get("tool_calls") or []
-            call_map: Dict[str, Dict[str, Any]] = {
-                str(c.get("id")): c
-                for c in tool_calls
-                if isinstance(c, dict) and isinstance(c.get("id"), str)
-            }
+        async with mcp_client_session(user, current_path) as mcp_session:
+            tools_schema = await _list_mcp_tools(mcp_session)
+            tool_index = {tool["name"]: tool for tool in tools_schema if tool.get("name")}
 
-            existing_ids = _existing_tool_result_ids(internal_messages)
-            for call_id in approved_ids | rejected_ids:
-                if call_id in existing_ids:
-                    continue
-                tool_call = call_map.get(call_id)
-                if not tool_call:
-                    continue
-                fn = tool_call.get("function") or {}
-                name = fn.get("name") if isinstance(fn, dict) else None
-                args_raw = fn.get("arguments") if isinstance(fn, dict) else None
-                args: Dict[str, Any] = {}
-                if isinstance(args_raw, str) and args_raw.strip():
-                    try:
-                        parsed = json.loads(args_raw)
-                        if isinstance(parsed, dict):
-                            args = parsed
-                    except json.JSONDecodeError:
-                        args = {}
-
-                spec = get_tool(str(name or ""))
-                if call_id in rejected_ids:
-                    content = tool_result_to_content({"canceled": True, "reason": "user_rejected"})
-                    tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                    internal_messages.append(tool_msg)
-                    new_messages.append(tool_msg)
-                    continue
-
-                if not spec:
-                    content = tool_result_to_content({"error": f"unknown_tool: {name}"})
-                    tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                    internal_messages.append(tool_msg)
-                    new_messages.append(tool_msg)
-                    continue
-
-                try:
-                    result = await spec.handler(args)
-                    content = tool_result_to_content(result)
-                except Exception as exc:  # noqa: BLE001
-                    content = tool_result_to_content({"error": str(exc)})
-                tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                internal_messages.append(tool_msg)
-                new_messages.append(tool_msg)
-
-        tools_schema = openai_tools()
-        ability = await _choose_chat_ability()
-        max_loops = 4
-
-        for _ in range(max_loops):
-            try:
-                assistant = await chat_completion(
-                    internal_messages,
-                    ability=ability,
-                    tools=tools_schema,
-                    tool_choice="auto",
-                    timeout=60.0,
-                )
-            except MissingModelError as exc:
-                raise HTTPException(status_code=400, detail=str(exc)) from exc
-            except httpx.HTTPStatusError as exc:
-                raise HTTPException(status_code=502, detail=f"对话请求失败: {exc}") from exc
-            except httpx.RequestError as exc:
-                raise HTTPException(status_code=502, detail=f"对话请求异常: {exc}") from exc
-
-            assistant = _ensure_tool_call_ids(assistant)
-            internal_messages.append(assistant)
-            new_messages.append(assistant)
-
-            tool_calls = assistant.get("tool_calls")
-            if not isinstance(tool_calls, list) or not tool_calls:
-                break
-
-            pending = []
-            for call in tool_calls:
-                if not isinstance(call, dict):
-                    continue
-                call_id = str(call.get("id") or "")
-                fn = call.get("function") or {}
-                name = fn.get("name") if isinstance(fn, dict) else None
-                args_raw = fn.get("arguments") if isinstance(fn, dict) else None
-                args: Dict[str, Any] = {}
-                if isinstance(args_raw, str) and args_raw.strip():
-                    try:
-                        parsed = json.loads(args_raw)
-                        if isinstance(parsed, dict):
-                            args = parsed
-                    except json.JSONDecodeError:
-                        args = {}
-
-                spec = get_tool(str(name or ""))
-                if not spec:
-                    content = tool_result_to_content({"error": f"unknown_tool: {name}"})
-                    tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                    internal_messages.append(tool_msg)
-                    new_messages.append(tool_msg)
-                    continue
-
-                if spec.requires_confirmation and not req.auto_execute:
-                    pending.append(_extract_pending(call, True))
-                    continue
-
-                try:
-                    result = await spec.handler(args)
-                    content = tool_result_to_content(result)
-                except Exception as exc:  # noqa: BLE001
-                    content = tool_result_to_content({"error": str(exc)})
-                tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                internal_messages.append(tool_msg)
-                new_messages.append(tool_msg)
-
-            if pending:
-                break
-
-        payload: Dict[str, Any] = {"messages": new_messages}
-        if pending:
-            payload["pending_tool_calls"] = [p.model_dump() for p in pending]
-        return payload
-
-    @classmethod
-    async def chat_stream(cls, req: AgentChatRequest, user: Optional[User]):
-        history: List[Dict[str, Any]] = list(req.messages or [])
-        current_path = _normalize_path(req.context.current_path if req.context else None)
-
-        system_prompt = _build_system_prompt(current_path)
-        internal_messages: List[Dict[str, Any]] = [{"role": "system", "content": system_prompt}] + history
-
-        new_messages: List[Dict[str, Any]] = []
-        pending: List[PendingToolCall] = []
-
-        approved_ids = {i for i in (req.approved_tool_call_ids or []) if isinstance(i, str) and i.strip()}
-        rejected_ids = {i for i in (req.rejected_tool_call_ids or []) if isinstance(i, str) and i.strip()}
-
-        try:
             if approved_ids or rejected_ids:
-                _, last_call_msg = _find_last_assistant_tool_calls(internal_messages)
-                last_call_msg = _ensure_tool_call_ids(last_call_msg)
-                tool_calls = last_call_msg.get("tool_calls") or []
+                _, last_call_msg = _find_last_assistant_mcp_calls(internal_messages)
+                last_call_msg = _ensure_mcp_call_ids(last_call_msg)
+                mcp_calls = last_call_msg.get("mcp_calls") or []
                 call_map: Dict[str, Dict[str, Any]] = {
-                    str(c.get("id")): c
-                    for c in tool_calls
-                    if isinstance(c, dict) and isinstance(c.get("id"), str)
+                    str(call.get("id")): call
+                    for call in mcp_calls
+                    if isinstance(call, dict) and isinstance(call.get("id"), str)
                 }
 
-                existing_ids = _existing_tool_result_ids(internal_messages)
+                existing_ids = _existing_mcp_result_ids(internal_messages)
                 for call_id in approved_ids | rejected_ids:
                     if call_id in existing_ids:
                         continue
-                    tool_call = call_map.get(call_id)
-                    if not tool_call:
+                    mcp_call = call_map.get(call_id)
+                    if not mcp_call:
                         continue
-                    fn = tool_call.get("function") or {}
-                    name = fn.get("name") if isinstance(fn, dict) else None
-                    args_raw = fn.get("arguments") if isinstance(fn, dict) else None
-                    args: Dict[str, Any] = {}
-                    if isinstance(args_raw, str) and args_raw.strip():
-                        try:
-                            parsed = json.loads(args_raw)
-                            if isinstance(parsed, dict):
-                                args = parsed
-                        except json.JSONDecodeError:
-                            args = {}
+                    name = str(mcp_call.get("name") or "")
+                    arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {}
+                    tool_desc = tool_index.get(name)
 
-                    spec = get_tool(str(name or ""))
                     if call_id in rejected_ids:
                         content = tool_result_to_content({"canceled": True, "reason": "user_rejected"})
-                        tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                        internal_messages.append(tool_msg)
-                        new_messages.append(tool_msg)
-                        yield _sse("tool_end", {"tool_call_id": call_id, "name": str(name or ""), "message": tool_msg})
-                        continue
-
-                    if not spec:
+                    elif not tool_desc:
                         content = tool_result_to_content({"error": f"unknown_tool: {name}"})
-                        tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
-                        internal_messages.append(tool_msg)
-                        new_messages.append(tool_msg)
-                        yield _sse("tool_end", {"tool_call_id": call_id, "name": str(name or ""), "message": tool_msg})
-                        continue
-
-                    yield _sse("tool_start", {"tool_call_id": call_id, "name": spec.name})
-                    try:
-                        result = await spec.handler(args)
-                        content = tool_result_to_content(result)
-                    except Exception as exc:  # noqa: BLE001
-                        content = tool_result_to_content({"error": str(exc)})
-                    tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
+                    else:
+                        try:
+                            content = await _execute_mcp_call(mcp_session, name, arguments)
+                        except Exception as exc:  # noqa: BLE001
+                            content = tool_result_to_content({"error": str(exc)})
+                    tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
                     internal_messages.append(tool_msg)
                     new_messages.append(tool_msg)
-                    yield _sse("tool_end", {"tool_call_id": call_id, "name": spec.name, "message": tool_msg})
 
-            tools_schema = openai_tools()
             ability = await _choose_chat_ability()
-            max_loops = 4
 
-            for _ in range(max_loops):
-                assistant_event_id = uuid.uuid4().hex
-                yield _sse("assistant_start", {"id": assistant_event_id})
-
-                assistant_message: Dict[str, Any] | None = None
+            for _ in range(8):
                 try:
-                    async for event in chat_completion_stream(
+                    assistant = await chat_completion(
                         internal_messages,
                         ability=ability,
                         tools=tools_schema,
                         tool_choice="auto",
                         timeout=60.0,
-                    ):
-                        if event.get("type") == "delta":
-                            delta = event.get("delta")
-                            if isinstance(delta, str) and delta:
-                                yield _sse("assistant_delta", {"id": assistant_event_id, "delta": delta})
-                        elif event.get("type") == "message":
-                            msg = event.get("message")
-                            if isinstance(msg, dict):
-                                assistant_message = msg
+                    )
                 except MissingModelError as exc:
                     raise HTTPException(status_code=400, detail=_format_exc(exc)) from exc
                 except httpx.HTTPStatusError as exc:
@@ -388,66 +231,196 @@ class AgentService:
                 except httpx.RequestError as exc:
                     raise HTTPException(status_code=502, detail=f"对话请求异常: {_format_exc(exc)}") from exc
 
-                if not assistant_message:
-                    assistant_message = {"role": "assistant", "content": ""}
+                assistant = _ensure_mcp_call_ids(assistant if isinstance(assistant, dict) else {"role": "assistant", "content": ""})
+                internal_messages.append(assistant)
+                new_messages.append(assistant)
 
-                assistant_message = _ensure_tool_call_ids(assistant_message)
-                internal_messages.append(assistant_message)
-                new_messages.append(assistant_message)
-                yield _sse("assistant_end", {"id": assistant_event_id, "message": assistant_message})
-
-                tool_calls = assistant_message.get("tool_calls")
-                if not isinstance(tool_calls, list) or not tool_calls:
+                mcp_calls = assistant.get("mcp_calls")
+                if not isinstance(mcp_calls, list) or not mcp_calls:
                     break
 
                 pending = []
-                for call in tool_calls:
+                for call in mcp_calls:
                     if not isinstance(call, dict):
                         continue
                     call_id = str(call.get("id") or "")
-                    fn = call.get("function") or {}
-                    name = fn.get("name") if isinstance(fn, dict) else None
-                    args_raw = fn.get("arguments") if isinstance(fn, dict) else None
-                    args: Dict[str, Any] = {}
-                    if isinstance(args_raw, str) and args_raw.strip():
-                        try:
-                            parsed = json.loads(args_raw)
-                            if isinstance(parsed, dict):
-                                args = parsed
-                        except json.JSONDecodeError:
-                            args = {}
+                    name = str(call.get("name") or "")
+                    arguments = call.get("arguments") if isinstance(call.get("arguments"), dict) else {}
+                    tool_desc = tool_index.get(name)
 
-                    spec = get_tool(str(name or ""))
-                    if not spec:
+                    if not tool_desc:
                         content = tool_result_to_content({"error": f"unknown_tool: {name}"})
-                        tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
+                        tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
                         internal_messages.append(tool_msg)
                         new_messages.append(tool_msg)
-                        yield _sse("tool_end", {"tool_call_id": call_id, "name": str(name or ""), "message": tool_msg})
                         continue
 
-                    if spec.requires_confirmation and not req.auto_execute:
+                    if _tool_requires_confirmation(tool_desc) and not req.auto_execute:
                         pending.append(_extract_pending(call, True))
                         continue
 
-                    yield _sse("tool_start", {"tool_call_id": call_id, "name": spec.name})
                     try:
-                        result = await spec.handler(args)
-                        content = tool_result_to_content(result)
+                        content = await _execute_mcp_call(mcp_session, name, arguments)
                     except Exception as exc:  # noqa: BLE001
                         content = tool_result_to_content({"error": str(exc)})
-                    tool_msg = {"role": "tool", "tool_call_id": call_id, "content": content}
+                    tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
                     internal_messages.append(tool_msg)
                     new_messages.append(tool_msg)
-                    yield _sse("tool_end", {"tool_call_id": call_id, "name": spec.name, "message": tool_msg})
 
                 if pending:
-                    yield _sse("pending", {"pending_tool_calls": [p.model_dump() for p in pending]})
                     break
 
+        payload: Dict[str, Any] = {"messages": new_messages}
+        if pending:
+            payload["pending_mcp_calls"] = [item.model_dump() for item in pending]
+        return payload
+
+    @classmethod
+    async def chat_stream(cls, req: AgentChatRequest, user: Optional[User]):
+        history: List[Dict[str, Any]] = list(req.messages or [])
+        current_path = _normalize_path(req.context.current_path if req.context else None)
+        system_prompt = _build_system_prompt(current_path)
+        internal_messages: List[Dict[str, Any]] = [{"role": "system", "content": system_prompt}] + history
+        new_messages: List[Dict[str, Any]] = []
+        pending: List[PendingMcpCall] = []
+
+        approved_ids = {i for i in (req.approved_mcp_call_ids or []) if isinstance(i, str) and i.strip()}
+        rejected_ids = {i for i in (req.rejected_mcp_call_ids or []) if isinstance(i, str) and i.strip()}
+
+        try:
+            async with mcp_client_session(user, current_path) as mcp_session:
+                tools_schema = await _list_mcp_tools(mcp_session)
+                tool_index = {tool["name"]: tool for tool in tools_schema if tool.get("name")}
+
+                if approved_ids or rejected_ids:
+                    _, last_call_msg = _find_last_assistant_mcp_calls(internal_messages)
+                    last_call_msg = _ensure_mcp_call_ids(last_call_msg)
+                    mcp_calls = last_call_msg.get("mcp_calls") or []
+                    call_map: Dict[str, Dict[str, Any]] = {
+                        str(call.get("id")): call
+                        for call in mcp_calls
+                        if isinstance(call, dict) and isinstance(call.get("id"), str)
+                    }
+
+                    existing_ids = _existing_mcp_result_ids(internal_messages)
+                    for call_id in approved_ids | rejected_ids:
+                        if call_id in existing_ids:
+                            continue
+                        mcp_call = call_map.get(call_id)
+                        if not mcp_call:
+                            continue
+
+                        name = str(mcp_call.get("name") or "")
+                        arguments = mcp_call.get("arguments") if isinstance(mcp_call.get("arguments"), dict) else {}
+                        tool_desc = tool_index.get(name)
+
+                        if call_id in rejected_ids:
+                            content = tool_result_to_content({"canceled": True, "reason": "user_rejected"})
+                            tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
+                            internal_messages.append(tool_msg)
+                            new_messages.append(tool_msg)
+                            yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg})
+                            continue
+
+                        if not tool_desc:
+                            content = tool_result_to_content({"error": f"unknown_tool: {name}"})
+                            tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
+                            internal_messages.append(tool_msg)
+                            new_messages.append(tool_msg)
+                            yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg})
+                            continue
+
+                        yield _sse("mcp_call_start", {"mcp_call_id": call_id, "name": name})
+                        try:
+                            content = await _execute_mcp_call(mcp_session, name, arguments)
+                        except Exception as exc:  # noqa: BLE001
+                            content = tool_result_to_content({"error": str(exc)})
+                        tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
+                        internal_messages.append(tool_msg)
+                        new_messages.append(tool_msg)
+                        yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg})
+
+                ability = await _choose_chat_ability()
+
+                for _ in range(8):
+                    assistant_event_id = str(uuid.uuid4())
+                    yield _sse("assistant_start", {"id": assistant_event_id})
+
+                    assistant_message: Dict[str, Any] | None = None
+                    try:
+                        async for event in chat_completion_stream(
+                            internal_messages,
+                            ability=ability,
+                            tools=tools_schema,
+                            tool_choice="auto",
+                            timeout=60.0,
+                        ):
+                            event_type = event.get("type")
+                            if event_type == "delta":
+                                delta = event.get("delta")
+                                if isinstance(delta, str) and delta:
+                                    yield _sse("assistant_delta", {"id": assistant_event_id, "delta": delta})
+                            elif event_type == "message":
+                                msg = event.get("message")
+                                if isinstance(msg, dict):
+                                    assistant_message = msg
+                    except MissingModelError as exc:
+                        raise HTTPException(status_code=400, detail=_format_exc(exc)) from exc
+                    except httpx.HTTPStatusError as exc:
+                        raise HTTPException(status_code=502, detail=f"对话请求失败: {_format_exc(exc)}") from exc
+                    except httpx.RequestError as exc:
+                        raise HTTPException(status_code=502, detail=f"对话请求异常: {_format_exc(exc)}") from exc
+
+                    if not assistant_message:
+                        assistant_message = {"role": "assistant", "content": ""}
+
+                    assistant_message = _ensure_mcp_call_ids(assistant_message)
+                    internal_messages.append(assistant_message)
+                    new_messages.append(assistant_message)
+                    yield _sse("assistant_end", {"id": assistant_event_id, "message": assistant_message})
+
+                    mcp_calls = assistant_message.get("mcp_calls")
+                    if not isinstance(mcp_calls, list) or not mcp_calls:
+                        break
+
+                    pending = []
+                    for call in mcp_calls:
+                        if not isinstance(call, dict):
+                            continue
+                        call_id = str(call.get("id") or "")
+                        name = str(call.get("name") or "")
+                        arguments = call.get("arguments") if isinstance(call.get("arguments"), dict) else {}
+                        tool_desc = tool_index.get(name)
+
+                        if not tool_desc:
+                            content = tool_result_to_content({"error": f"unknown_tool: {name}"})
+                            tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
+                            internal_messages.append(tool_msg)
+                            new_messages.append(tool_msg)
+                            yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg})
+                            continue
+
+                        if _tool_requires_confirmation(tool_desc) and not req.auto_execute:
+                            pending.append(_extract_pending(call, True))
+                            continue
+
+                        yield _sse("mcp_call_start", {"mcp_call_id": call_id, "name": name})
+                        try:
+                            content = await _execute_mcp_call(mcp_session, name, arguments)
+                        except Exception as exc:  # noqa: BLE001
+                            content = tool_result_to_content({"error": str(exc)})
+                        tool_msg = {"role": "tool", "mcp_call_id": call_id, "content": content}
+                        internal_messages.append(tool_msg)
+                        new_messages.append(tool_msg)
+                        yield _sse("mcp_call_end", {"mcp_call_id": call_id, "name": name, "message": tool_msg})
+
+                    if pending:
+                        yield _sse("pending", {"pending_mcp_calls": [item.model_dump() for item in pending]})
+                        break
+
             payload: Dict[str, Any] = {"messages": new_messages}
             if pending:
-                payload["pending_tool_calls"] = [p.model_dump() for p in pending]
+                payload["pending_mcp_calls"] = [item.model_dump() for item in pending]
             yield _sse("done", payload)
 
         except asyncio.CancelledError:
@@ -460,13 +433,11 @@ class AgentService:
             new_messages.append({"role": "assistant", "content": content})
             payload: Dict[str, Any] = {"messages": new_messages}
             if pending:
-                payload["pending_tool_calls"] = [p.model_dump() for p in pending]
+                payload["pending_mcp_calls"] = [item.model_dump() for item in pending]
             yield _sse("done", payload)
-            return
         except Exception as exc:  # noqa: BLE001
             new_messages.append({"role": "assistant", "content": f"服务端异常: {_format_exc(exc)}"})
             payload: Dict[str, Any] = {"messages": new_messages}
             if pending:
-                payload["pending_tool_calls"] = [p.model_dump() for p in pending]
+                payload["pending_mcp_calls"] = [item.model_dump() for item in pending]
             yield _sse("done", payload)
-            return

domain/agent/tools/__init__.py

@@ -1,6 +1,6 @@
 from typing import Any, Dict, List, Optional
 
-from .base import ToolSpec, tool_result_to_content
+from .base import McpToolDescriptor, ToolSpec, tool_result_to_content, tool_spec_to_mcp_descriptor
 from .processors import TOOLS as PROCESSOR_TOOLS
 from .time import TOOLS as TIME_TOOLS
 from .vfs import TOOLS as VFS_TOOLS
@@ -15,23 +15,19 @@ def get_tool(name: str) -> Optional[ToolSpec]:
     return TOOLS.get(name)
 
 
-def openai_tools() -> List[Dict[str, Any]]:
-    out: List[Dict[str, Any]] = []
-    for spec in TOOLS.values():
-        out.append({
-            "type": "function",
-            "function": {
-                "name": spec.name,
-                "description": spec.description,
-                "parameters": spec.parameters,
-            },
-        })
-    return out
+def list_tool_specs() -> List[ToolSpec]:
+    return list(TOOLS.values())
+
+
+def mcp_tool_descriptors() -> List[McpToolDescriptor]:
+    return [tool_spec_to_mcp_descriptor(spec) for spec in TOOLS.values()]
 
 
 __all__ = [
+    "McpToolDescriptor",
     "ToolSpec",
     "get_tool",
-    "openai_tools",
+    "list_tool_specs",
+    "mcp_tool_descriptors",
     "tool_result_to_content",
 ]

domain/agent/tools/base.py

@@ -3,6 +3,16 @@ from dataclasses import dataclass
 from typing import Any, Awaitable, Callable, Dict, List, Optional
 
 
+@dataclass(frozen=True)
+class McpToolDescriptor:
+    name: str
+    description: str
+    input_schema: Dict[str, Any]
+    annotations: Dict[str, Any]
+    meta: Dict[str, Any]
+    requires_confirmation: bool
+
+
 @dataclass(frozen=True)
 class ToolSpec:
     name: str
@@ -141,9 +151,31 @@ def _normalize_tool_result(result: Any) -> Dict[str, Any]:
     return {"ok": True, "summary": summary, "view": view, "data": result}
 
 
+def normalize_tool_result(result: Any) -> Dict[str, Any]:
+    return _normalize_tool_result(result)
+
+
 def tool_result_to_content(result: Any) -> str:
-    payload = _normalize_tool_result(result)
+    payload = normalize_tool_result(result)
     try:
         return json.dumps(payload, ensure_ascii=False, default=str)
     except TypeError:
         return json.dumps({"ok": False, "summary": "error", "view": {"type": "error", "message": "error"}}, ensure_ascii=False)
+
+
+def tool_spec_to_mcp_descriptor(spec: ToolSpec) -> McpToolDescriptor:
+    read_only = not spec.requires_confirmation
+    annotations: Dict[str, Any] = {
+        "readOnlyHint": read_only,
+        "destructiveHint": bool(spec.requires_confirmation),
+    }
+    if spec.name == "web_fetch":
+        annotations["openWorldHint"] = True
+    return McpToolDescriptor(
+        name=spec.name,
+        description=spec.description,
+        input_schema=spec.parameters,
+        annotations=annotations,
+        meta={"requires_confirmation": spec.requires_confirmation},
+        requires_confirmation=spec.requires_confirmation,
+    )

domain/agent/types.py

@@ -10,14 +10,19 @@ class AgentChatContext(BaseModel):
 class AgentChatRequest(BaseModel):
     messages: List[Dict[str, Any]] = Field(default_factory=list)
     auto_execute: bool = False
-    approved_tool_call_ids: List[str] = Field(default_factory=list)
-    rejected_tool_call_ids: List[str] = Field(default_factory=list)
+    approved_mcp_call_ids: List[str] = Field(default_factory=list)
+    rejected_mcp_call_ids: List[str] = Field(default_factory=list)
     context: Optional[AgentChatContext] = None
 
 
-class PendingToolCall(BaseModel):
+class McpCall(BaseModel):
+    id: str
+    name: str
+    arguments: Dict[str, Any] = Field(default_factory=dict)
+
+
+class PendingMcpCall(BaseModel):
     id: str
     name: str
     arguments: Dict[str, Any] = Field(default_factory=dict)
     requires_confirmation: bool = True
-

domain/ai/inference.py

@@ -15,6 +15,102 @@ class MissingModelError(RuntimeError):
     pass
 
 
+def _mcp_tools_to_openai_wire(tools: List[Dict[str, Any]] | None) -> List[Dict[str, Any]] | None:
+    if not tools:
+        return None
+    out: List[Dict[str, Any]] = []
+    for tool in tools:
+        if not isinstance(tool, dict):
+            continue
+        name = tool.get("name")
+        if not isinstance(name, str) or not name.strip():
+            continue
+        out.append(
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": str(tool.get("description") or ""),
+                    "parameters": tool.get("input_schema") if isinstance(tool.get("input_schema"), dict) else {},
+                },
+            }
+        )
+    return out
+
+
+def _mcp_messages_to_openai_wire(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
+    for message in messages:
+        if not isinstance(message, dict):
+            continue
+        item = dict(message)
+        mcp_call_id = item.pop("mcp_call_id", None)
+        if isinstance(mcp_call_id, str) and mcp_call_id.strip():
+            item["tool_call_id"] = mcp_call_id
+
+        mcp_calls = item.pop("mcp_calls", None)
+        if isinstance(mcp_calls, list):
+            tool_calls: List[Dict[str, Any]] = []
+            for idx, call in enumerate(mcp_calls):
+                if not isinstance(call, dict):
+                    continue
+                name = call.get("name")
+                if not isinstance(name, str) or not name.strip():
+                    continue
+                arguments = call.get("arguments") if isinstance(call.get("arguments"), dict) else {}
+                tool_calls.append(
+                    {
+                        "id": str(call.get("id") or f"call_{idx}"),
+                        "type": "function",
+                        "function": {
+                            "name": name,
+                            "arguments": json.dumps(arguments, ensure_ascii=False),
+                        },
+                    }
+                )
+            if tool_calls:
+                item["tool_calls"] = tool_calls
+        out.append(item)
+    return out
+
+
+def _openai_wire_message_to_mcp(message: Dict[str, Any]) -> Dict[str, Any]:
+    out = dict(message)
+    tool_call_id = out.pop("tool_call_id", None)
+    if isinstance(tool_call_id, str) and tool_call_id.strip():
+        out["mcp_call_id"] = tool_call_id
+
+    tool_calls = out.pop("tool_calls", None)
+    if isinstance(tool_calls, list):
+        mcp_calls: List[Dict[str, Any]] = []
+        for idx, call in enumerate(tool_calls):
+            if not isinstance(call, dict):
+                continue
+            fn = call.get("function") if isinstance(call.get("function"), dict) else {}
+            name = fn.get("name")
+            if not isinstance(name, str) or not name.strip():
+                continue
+            arguments: Dict[str, Any] = {}
+            raw_args = fn.get("arguments")
+            if isinstance(raw_args, str) and raw_args.strip():
+                try:
+                    parsed = json.loads(raw_args)
+                    if isinstance(parsed, dict):
+                        arguments = parsed
+                except json.JSONDecodeError:
+                    arguments = {}
+            mcp_calls.append(
+                {
+                    "id": str(call.get("id") or f"call_{idx}"),
+                    "name": name,
+                    "arguments": arguments,
+                }
+            )
+        if mcp_calls:
+            out["mcp_calls"] = mcp_calls
+    return out
+
+
 async def describe_image_base64(base64_image: str, detail: str = "high") -> str:
     """
     传入 base64 图片并返回描述文本。缺省时返回错误提示。
@@ -939,34 +1035,39 @@ async def chat_completion(
 ) -> Dict[str, Any]:
     model, provider = await _require_model(ability)
     fmt = str(provider.api_format or "").lower()
+    wire_messages = _mcp_messages_to_openai_wire(messages)
+    wire_tools = _mcp_tools_to_openai_wire(tools)
     if fmt == "openai":
-        return await _chat_with_openai(
+        result = await _chat_with_openai(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             tool_choice=tool_choice,
             temperature=temperature,
             timeout=timeout,
         )
+        return _openai_wire_message_to_mcp(result)
     if fmt == "anthropic":
-        return await _chat_with_anthropic(
+        result = await _chat_with_anthropic(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             temperature=temperature,
             timeout=timeout,
         )
+        return _openai_wire_message_to_mcp(result)
     if fmt == "ollama":
-        return await _chat_with_ollama(
+        result = await _chat_with_ollama(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             temperature=temperature,
             timeout=timeout,
         )
+        return _openai_wire_message_to_mcp(result)
     raise MissingModelError(f"当前不支持该对话模型接口类型: {provider.api_format}")
 
 
@@ -1016,38 +1117,49 @@ async def chat_completion_stream(
 ) -> AsyncIterator[Dict[str, Any]]:
     model, provider = await _require_model(ability)
     fmt = str(provider.api_format or "").lower()
+    wire_messages = _mcp_messages_to_openai_wire(messages)
+    wire_tools = _mcp_tools_to_openai_wire(tools)
     if fmt == "openai":
         async for event in _chat_stream_with_openai(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             tool_choice=tool_choice,
             temperature=temperature,
             timeout=timeout,
         ):
+            if event.get("type") == "message" and isinstance(event.get("message"), dict):
+                yield {**event, "message": _openai_wire_message_to_mcp(event["message"])}
+                continue
             yield event
         return
     if fmt == "anthropic":
         async for event in _chat_stream_with_anthropic(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             temperature=temperature,
             timeout=timeout,
         ):
+            if event.get("type") == "message" and isinstance(event.get("message"), dict):
+                yield {**event, "message": _openai_wire_message_to_mcp(event["message"])}
+                continue
             yield event
         return
     if fmt == "ollama":
         async for event in _chat_stream_with_ollama(
             provider,
             model,
-            messages,
-            tools=tools,
+            wire_messages,
+            tools=wire_tools,
             temperature=temperature,
             timeout=timeout,
         ):
+            if event.get("type") == "message" and isinstance(event.get("message"), dict):
+                yield {**event, "message": _openai_wire_message_to_mcp(event["message"])}
+                continue
             yield event
         return
     raise MissingModelError(f"当前不支持该对话模型接口类型: {provider.api_format}")