diff --git a/docs/ai-chat-api-design.md b/docs/ai-chat-api-design.md
index ecd099651..cee64bd25 100644
--- a/docs/ai-chat-api-design.md
+++ b/docs/ai-chat-api-design.md
@@ -179,6 +179,8 @@ GNS3 Copilot Agent 需要以下信息才能正常工作：
 - Token 数据依赖 LLM 返回的 `usage_metadata`，某些模型可能不支持
 - 统计数据在流结束后通过 `update_session` 方法增量更新到数据库
 - LangGraph 已自动处理 input 和 output 的历史累加，代码使用最后一次 LLM 调用的值
+- **消息 ID 处理**：创建初始消息时分配 ID（`HumanMessage(id=str(uuid4()))`），从 checkpoint 读取的消息如果没有 ID 也会自动生成
+- **格式转换**：使用 `message_converters.py` 模块处理 LangChain 和 OpenAI 格式之间的转换，确保 tool_calls 格式符合 OpenAI 规范
 
 ### Title 自动同步
 
@@ -348,16 +350,55 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。
 
 ### OpenAIMessage
 
-- id: str - 消息 ID
-- role: Literal["user", "assistant", "system", "tool"] - 角色
-- content: str - 消息内容
-- name: Optional[str] - 工具消息名称
-- tool_call_id: Optional[str] - 关联的工具调用 ID
-- tool_calls: Optional[List] - 工具调用列表（assistant 消息）
-- created_at: str - 创建时间
+OpenAI 兼容的消息模型。
+
+**基础字段**：
+- id: str - 消息唯一标识符（自动生成或从 LangChain 消息继承）
+- role: Literal["user", "assistant", "system", "tool"] - 消息角色
+- content: str - 消息内容（支持文本、JSON 字符串）
+- created_at: str - 创建时间（ISO 8601）
+
+**工具相关字段**：
+- name: Optional[str] - 工具消息名称（tool 消息）
+- tool_call_id: Optional[str] - 关联的工具调用 ID（tool 消息）
+- tool_calls: Optional[List[OpenAIToolCall]] - 工具调用列表（assistant 消息）
+  - id: str - 工具调用 ID
+  - type: Literal["function"] - 固定为 "function"
+  - function: Dict - 包含 name 和 arguments（dict 或 JSON 字符串）
+
+**元数据**：
+- metadata: Optional[Dict] - 额外的消息元数据
 
 ## 核心组件
 
+### Message Converters（消息格式转换）
+
+**文件**：`gns3server/agent/gns3_copilot/utils/message_converters.py`
+
+**职责**：在 LangChain 消息格式和 OpenAI 兼容格式之间进行转换
+
+**主要函数**：
+- `convert_langchain_to_openai()`：LangChain → OpenAI 格式
+- `convert_openai_to_langchain()`：OpenAI → LangChain 格式
+- `convert_stream_event_to_openai()`：流事件 → OpenAI SSE 格式
+
+**关键转换逻辑**：
+
+1. **消息 ID 处理**
+   - 如果消息没有 ID，自动生成 UUID
+   - 确保所有返回的消息都有唯一标识符
+
+2. **Tool Calls 格式转换**
+   - LangChain 格式：`{'name': 'xxx', 'args': {...}, 'id': 'yyy', 'type': 'tool_call'}`
+   - OpenAI 格式：`{'id': 'yyy', 'type': 'function', 'function': {'name': 'xxx', 'arguments': '{...}'}}`
+   - 自动将 `args` 对象转换为 JSON 字符串（如需要）
+
+3. **Content 类型处理**
+   - 支持 string、dict、list 类型
+   - 非 string 类型自动转换为 JSON 字符串
+
+**实现位置**：`utils/message_converters.py`
+
 ### AgentService
 
 **职责**：项目级的 Agent 管理服务
@@ -375,9 +416,10 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。
 2. 获取或创建 chat session（从 `chat_sessions` 表）
 3. 设置 ContextVars（JWT token、LLM config）
 4. 构建 LangGraph config
-5. 流式执行 Agent，同时收集统计信息
-6. 流结束后更新会话统计到数据库
-7. 同步 auto-generated title
+5. 创建带 ID 的初始消息：`HumanMessage(content=message, id=str(uuid4()))`
+6. 流式执行 Agent，同时收集统计信息
+7. 流结束后更新会话统计到数据库
+8. 同步 auto-generated title
 
 **统计收集机制**（在 `stream_chat` 中）：
 
@@ -387,10 +429,10 @@ Chat API 使用 Server-Sent Events (SSE) 进行流式传输。
 
 **关键事件处理**：
 - `on_chat_model_start`：LLM 调用次数 +1
-- `on_chat_model_end`：提取 token 使用量，AI 消息计数 +1
+- `on_chat_model_end`：提取 token 使用量（从 `output.usage_metadata`），AI 消息计数 +1
 - `on_tool_end`：工具消息计数 +1
 
-**实现位置**：`agent_service.py` 第 233-294 行
+**实现位置**：`agent_service.py`
 
 ### ProjectAgentManager
 
diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py
index 40292d843..091a38f62 100644
--- a/gns3server/agent/gns3_copilot/agent_service.py
+++ b/gns3server/agent/gns3_copilot/agent_service.py
@@ -20,6 +20,7 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
 from gns3server.agent.gns3_copilot.agent.gns3_copilot import agent_builder
 from gns3server.agent.gns3_copilot.chat_sessions_repository import ChatSessionsRepository
+from gns3server.agent.gns3_copilot.utils.message_converters import convert_langchain_to_openai
 
 log = logging.getLogger(__name__)
 
@@ -220,7 +221,7 @@ class AgentService:
 
         # Build inputs
         inputs = {
-            "messages": [HumanMessage(content=message)],
+            "messages": [HumanMessage(content=message, id=str(uuid4()))],
             "llm_calls": 0,
             "remaining_steps": 20,
             "mode": mode,
@@ -413,32 +414,8 @@ class AgentService:
         }
 
     def _convert_message_to_dict(self, msg) -> Dict[str, Any]:
-        """Convert a LangChain message to dict format."""
-        from datetime import datetime
-
-        msg_type = type(msg).__name__
-
-        result = {
-            "id": getattr(msg, "id", str(uuid4())),
-            "role": "user",
-            "content": getattr(msg, "content", str(msg)),
-            "created_at": datetime.utcnow().isoformat() + "Z",
-        }
-
-        if msg_type == "HumanMessage":
-            result["role"] = "user"
-        elif msg_type == "AIMessage":
-            result["role"] = "assistant"
-            if hasattr(msg, "tool_calls") and msg.tool_calls:
-                result["tool_calls"] = msg.tool_calls
-        elif msg_type == "ToolMessage":
-            result["role"] = "tool"
-            result["tool_call_id"] = getattr(msg, "tool_call_id", None)
-            result["name"] = getattr(msg, "name", None)
-        elif msg_type == "SystemMessage":
-            result["role"] = "system"
-
-        return result
+        """Convert a LangChain message to OpenAI-compatible dict format."""
+        return convert_langchain_to_openai(msg)
 
     async def list_sessions(self, user_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]:
         """
diff --git a/gns3server/agent/gns3_copilot/utils/message_converters.py b/gns3server/agent/gns3_copilot/utils/message_converters.py
new file mode 100644
index 000000000..dffc6dce2
--- /dev/null
+++ b/gns3server/agent/gns3_copilot/utils/message_converters.py
@@ -0,0 +1,240 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+# This file is part of GNS3 Server.
+#
+# GNS3 Server is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Message format converters for OpenAI-compatible message format.
+Converts between LangChain messages and OpenAI-compatible format.
+"""
+import json
+import uuid
+from datetime import datetime
+from typing import Dict, Any
+
+
+def _ensure_string(content: Any) -> str:
+    """Ensure content is a string, converting dicts/lists to JSON if needed."""
+    if isinstance(content, str):
+        return content
+    elif isinstance(content, (dict, list)):
+        return json.dumps(content, ensure_ascii=False, indent=2)
+    else:
+        return str(content)
+
+
+def convert_langchain_to_openai(lc_message) -> Dict[str, Any]:
+    """
+    Convert LangChain message to OpenAI-compatible format.
+
+    Args:
+        lc_message: LangChain message (HumanMessage, AIMessage, SystemMessage, ToolMessage)
+
+    Returns:
+        Dictionary in OpenAI-compatible format
+    """
+    from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
+
+    # Generate message ID
+    msg_id = getattr(lc_message, 'id', None)
+    if msg_id is None:
+        msg_id = str(uuid.uuid4())
+
+    # Get timestamp
+    timestamp = getattr(lc_message, 'created_at', None)
+    if timestamp is None:
+        timestamp = datetime.utcnow().isoformat()
+    elif hasattr(timestamp, 'isoformat'):
+        timestamp = timestamp.isoformat()
+
+    # Base message structure
+    base_msg = {
+        "id": msg_id,
+        "created_at": timestamp,
+        "metadata": {}
+    }
+
+    # Convert based on message type
+    if isinstance(lc_message, HumanMessage):
+        return {
+            **base_msg,
+            "role": "user",
+            "content": lc_message.content
+        }
+
+    elif isinstance(lc_message, AIMessage):
+        msg = {
+            **base_msg,
+            "role": "assistant",
+            "content": lc_message.content
+        }
+
+        # Handle tool calls - convert to OpenAI format
+        if hasattr(lc_message, 'tool_calls') and lc_message.tool_calls:
+            tool_calls = []
+            for tc in lc_message.tool_calls:
+                # Convert to dict if it's an object
+                tc_dict = tc if isinstance(tc, dict) else tc.model_dump()
+                tool_calls.append({
+                    "id": tc_dict.get("id", str(uuid.uuid4())),
+                    "type": "function",
+                    "function": {
+                        "name": tc_dict.get("name", ""),
+                        "arguments": tc_dict.get("args", {})
+                    }
+                })
+            msg["tool_calls"] = tool_calls
+
+        return msg
+
+    elif isinstance(lc_message, ToolMessage):
+        return {
+            **base_msg,
+            "role": "tool",
+            "content": _ensure_string(lc_message.content),
+            "name": getattr(lc_message, 'name', ''),
+            "tool_call_id": getattr(lc_message, 'tool_call_id', '')
+        }
+
+    elif isinstance(lc_message, SystemMessage):
+        return {
+            **base_msg,
+            "role": "system",
+            "content": lc_message.content
+        }
+
+    else:
+        # Fallback for unknown message types
+        return {
+            **base_msg,
+            "role": "unknown",
+            "content": str(lc_message)
+        }
+
+
+def convert_openai_to_langchain(msg: Dict[str, Any]):
+    """
+    Convert OpenAI-compatible format to LangChain message.
+
+    Args:
+        msg: Dictionary in OpenAI-compatible format
+
+    Returns:
+        LangChain message
+    """
+    from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage
+
+    role = msg.get("role", "user")
+    content = msg.get("content", "")
+
+    if role == "user":
+        return HumanMessage(content=content, id=msg.get("id"))
+
+    elif role == "assistant":
+        ai_msg = AIMessage(content=content, id=msg.get("id"))
+
+        # Restore tool calls if present
+        if "tool_calls" in msg and msg["tool_calls"]:
+            tool_calls = []
+            for tc in msg["tool_calls"]:
+                tool_calls.append({
+                    "id": tc.get("id", str(uuid.uuid4())),
+                    "name": tc.get("function", {}).get("name", ""),
+                    "args": tc.get("function", {}).get("arguments", {})
+                })
+            ai_msg.tool_calls = tool_calls
+
+        return ai_msg
+
+    elif role == "tool":
+        return ToolMessage(
+            content=content,
+            name=msg.get("name", ""),
+            tool_call_id=msg.get("tool_call_id", "")
+        )
+
+    elif role == "system":
+        return SystemMessage(content=content)
+
+    else:
+        # Fallback to HumanMessage for unknown roles
+        return HumanMessage(content=content)
+
+
+def convert_stream_event_to_openai(event: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Convert LangGraph streaming event to OpenAI-compatible format.
+
+    Args:
+        event: LangGraph streaming event
+
+    Returns:
+        Dictionary in OpenAI-compatible streaming response format
+    """
+    event_type = event.get("event", "")
+
+    if event_type == "on_chat_model_stream":
+        chunk = event.get("data", {}).get("chunk", {})
+        content = getattr(chunk, 'content', '')
+
+        if content:
+            return {
+                "type": "content",
+                "content": content,
+                "message_id": event.get("metadata", {}).get("msg_id")
+            }
+
+        # Check for tool call chunks
+        if hasattr(chunk, 'tool_call_chunks') and chunk.tool_call_chunks:
+            for tc_chunk in chunk.tool_call_chunks:
+                tc_id = getattr(tc_chunk, 'id', None)
+                tc_name = getattr(tc_chunk, 'name', None)
+                tc_args = getattr(tc_chunk, 'args', None)
+
+                if tc_id:
+                    return {
+                        "type": "tool_call",
+                        "tool_call": {
+                            "id": tc_id,
+                            "type": "function",
+                            "function": {
+                                "name": tc_name or "",
+                                "arguments": tc_args or ""
+                            }
+                        }
+                    }
+
+    elif event_type == "on_tool_start":
+        return {
+            "type": "tool_start",
+            "tool_name": event.get("name", ""),
+            "metadata": event.get("metadata", {})
+        }
+
+    elif event_type == "on_tool_end":
+        tool_output = event.get("data", {}).get("output", "")
+        # Convert dict or list output to JSON string for serialization
+        if isinstance(tool_output, (dict, list)):
+            tool_output = json.dumps(tool_output, ensure_ascii=False, indent=2)
+
+        return {
+            "type": "tool_end",
+            "tool_output": tool_output,
+            "tool_name": event.get("name", ""),
+            "metadata": event.get("metadata", {})
+        }
+
+    # Default empty response
+    return {"type": "unknown"}