diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
index e378aebde..f8271a287 100644
--- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
+++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py
@@ -124,9 +124,14 @@ class MessagesState(TypedDict):
# Define llm call node
-def llm_call(state: dict):
+def llm_call(state: dict, config: dict = None):
"""LLM decides whether to call a tool or not"""
+ # Extract user authentication info from LangGraph config
+ configurable = config.get("configurable", {}) if config else {}
+ user_id = configurable.get("user_id")
+ jwt_token = configurable.get("jwt_token")
+
# Defensive check: skip LLM call if no user messages
messages = state.get("messages", [])
if not messages or len(messages) == 0:
@@ -203,23 +208,35 @@ def llm_call(state: dict):
# print(full_messages)
# Create fresh model with tools for each LLM call
- # This ensures configuration changes in .env take effect immediately
- model_with_tools = create_base_model_with_tools(tools)
+ # This ensures configuration changes take effect immediately
+ # Pass user_id and jwt_token for per-user LLM config and API authentication
+ model_with_tools = create_base_model_with_tools(
+ tools,
+ user_id=user_id,
+ jwt_token=jwt_token
+ )
+ # Store jwt_token in state for Tools to use when calling GNS3 API
return {
"messages": [model_with_tools.invoke(full_messages)],
"llm_calls": state.get("llm_calls", 0) + 1,
"topology_info": topology_info,
+ "jwt_token": jwt_token,
}
# Define generate title node
-def generate_title(state: MessagesState) -> dict:
+def generate_title(state: MessagesState, config: dict = None) -> dict:
"""
Generate a conversation title using a lightweight assistant LLM (title_model).
This node is only executed when no title has been set yet (first round only).
"""
+ # Extract user authentication info from LangGraph config
+ configurable = config.get("configurable", {}) if config else {}
+ user_id = configurable.get("user_id")
+ jwt_token = configurable.get("jwt_token")
+
# Only generate a title if it hasn't been set yet
current_title = state.get("conversation_title")
if current_title in [None, "New Conversation"]:
@@ -236,7 +253,8 @@ def generate_title(state: MessagesState) -> dict:
# Call the title generation model (create fresh instance for each call)
try:
# Create fresh title model instance from current env configuration
- title_model = create_title_model()
+ # Pass user_id and jwt_token for per-user LLM config
+ title_model = create_title_model(user_id=user_id, jwt_token=jwt_token)
response = title_model.invoke(
title_prompt_messages, config={"configurable": {"foo_temperature": 1.0}}
)
diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py
new file mode 100644
index 000000000..c1c288771
--- /dev/null
+++ b/gns3server/agent/gns3_copilot/agent_service.py
@@ -0,0 +1,282 @@
+"""
+GNS3 Copilot Agent Service
+
+Provides project-level Agent instances with SQLite checkpoint management.
+Each project has its own AgentService with a dedicated checkpoint database
+in the project directory.
+"""
+
+import asyncio
+import logging
+import os
+from typing import AsyncGenerator, Dict, Any, Optional
+from uuid import uuid4
+
+import aiosqlite
+from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+from gns3_copilot.agent.gns3_copilot import agent_builder
+
+log = logging.getLogger(__name__)
+
+
+class AgentService:
+ """
+ Project-level Agent Service with async checkpoint management.
+
+ Manages a LangGraph agent instance with SQLite-based state persistence
+ for a single GNS3 project.
+ """
+
+ def __init__(self, project_path: str):
+ """
+ Initialize AgentService for a project.
+
+ Args:
+ project_path: Path to the GNS3 project directory
+ """
+ self.project_path = project_path
+ self._checkpointer: Optional[AsyncSqliteSaver] = None
+ self._checkpointer_conn: Optional[aiosqlite.Connection] = None
+ self._checkpointer_path: Optional[str] = None
+ self._graph = None
+ self._init_lock = asyncio.Lock()
+ self._initialized = False
+
+ def _get_checkpoint_dir(self) -> str:
+ """Get or create the checkpoint directory for this project."""
+ checkpoint_dir = os.path.join(self.project_path, ".gns3-copilot")
+ os.makedirs(checkpoint_dir, exist_ok=True)
+ return checkpoint_dir
+
+ async def _get_checkpointer(self) -> AsyncSqliteSaver:
+ """
+ Get or create the SQLite checkpointer for this project.
+
+ Returns:
+ AsyncSqliteSaver instance
+ """
+ async with self._init_lock:
+ if self._checkpointer is not None:
+ return self._checkpointer
+
+ checkpoint_dir = self._get_checkpoint_dir()
+ checkpointer_path = os.path.join(checkpoint_dir, "copilot_checkpoints.db")
+
+ log.debug("Creating checkpointer at: %s", checkpointer_path)
+
+ # Close existing connection if switching projects
+ if self._checkpointer_conn:
+ try:
+ await self._checkpointer_conn.close()
+ log.debug("Closed previous checkpointer connection")
+ except Exception as e:
+ log.warning("Error closing old checkpointer connection: %s", e)
+
+ # Create new connection
+ conn = await aiosqlite.connect(checkpointer_path)
+ # Enable WAL mode for better concurrent performance
+ await conn.execute("PRAGMA journal_mode=WAL;")
+ self._checkpointer_conn = conn # Save connection reference to prevent GC
+ self._checkpointer = AsyncSqliteSaver(conn)
+
+ # CRITICAL: Initialize database schema
+ await self._checkpointer.setup()
+
+ self._checkpointer_path = checkpointer_path
+ self._initialized = True
+
+ log.info("Project checkpointer created at: %s", checkpointer_path)
+ return self._checkpointer
+
+ async def _get_graph(self):
+ """Get or compile the LangGraph agent."""
+ if self._graph is None:
+ checkpointer = await self._get_checkpointer()
+ self._graph = agent_builder.compile(checkpointer=checkpointer)
+ log.info("LangGraph agent compiled for project: %s", self.project_path)
+ return self._graph
+
+ async def stream_chat(
+ self,
+ message: str,
+ session_id: str,
+ project_id: Optional[str] = None,
+ user_id: Optional[str] = None,
+ jwt_token: Optional[str] = None,
+ mode: str = "text"
+ ) -> AsyncGenerator[Dict[str, Any], None]:
+ """
+ Stream chat responses from the agent.
+
+ Args:
+ message: User message
+ session_id: Session/thread ID for conversation continuity
+ project_id: GNS3 project ID (optional, for context)
+ user_id: User ID for LLM config lookup (optional)
+ jwt_token: JWT token for API authentication (optional)
+ mode: Interaction mode (default: "text")
+
+ Yields:
+ Dict containing SSE-compatible response chunks
+ """
+ # Build config with user authentication info
+ config = {
+ "configurable": {
+ "thread_id": session_id,
+ "user_id": user_id,
+ "jwt_token": jwt_token,
+ }
+ }
+
+ # Build inputs
+ inputs = {
+ "messages": [HumanMessage(content=message)],
+ "llm_calls": 0,
+ "remaining_steps": 20,
+ "mode": mode,
+ }
+
+ # Get the compiled graph
+ graph = await self._get_graph()
+
+ # Stream events
+ try:
+ async for event in graph.astream_events(inputs, config=config, version="v2"):
+ chunk = self._convert_event_to_chunk(event, session_id)
+ if chunk:
+ yield chunk
+
+ yield {"type": "done", "session_id": session_id}
+
+ except Exception as e:
+ log.error("Error in stream_chat: %s", e, exc_info=True)
+ yield {"type": "error", "error": str(e), "session_id": session_id}
+
+ def _convert_event_to_chunk(self, event: Dict[str, Any], session_id: str) -> Optional[Dict[str, Any]]:
+ """
+ Convert LangGraph event to API response chunk.
+
+ Args:
+ event: LangGraph event from astream_events
+ session_id: Session ID for the response
+
+ Returns:
+ Dict for SSE response or None if event should be filtered
+ """
+ event_type = event.get("event", "")
+ data = event.get("data", {})
+
+ if event_type == "on_chat_model_stream":
+ # Streaming text content from LLM
+ chunk = data.get("chunk", {})
+ content = chunk.get("content", "")
+ if content:
+ return {"type": "content", "content": content}
+
+ elif event_type == "on_tool_start":
+ # Tool execution started
+ return {
+ "type": "tool_start",
+ "tool_name": event.get("name", ""),
+ "session_id": session_id
+ }
+
+ elif event_type == "on_tool_end":
+ # Tool execution completed
+ output = data.get("output", "")
+ # Convert output to string if it's not already
+ if not isinstance(output, str):
+ output = str(output)
+ return {
+ "type": "tool_end",
+ "tool_name": event.get("name", ""),
+ "tool_output": output,
+ "session_id": session_id
+ }
+
+ return None
+
+ async def get_history(self, session_id: str, limit: int = 100) -> Dict[str, Any]:
+ """
+ Get conversation history for a session.
+
+ Args:
+ session_id: Session/thread ID
+ limit: Maximum number of messages to retrieve
+
+ Returns:
+ Dict containing thread_id, title, and messages
+ """
+ config = {"configurable": {"thread_id": session_id}}
+
+ try:
+ graph = await self._get_graph()
+ state = await graph.aget_state(config)
+
+ if state and "messages" in state.values:
+ messages = []
+ for msg in state.values["messages"][-limit:]:
+ messages.append(self._convert_message_to_dict(msg))
+
+ title = state.values.get("conversation_title", "New Conversation")
+
+ return {
+ "thread_id": session_id,
+ "title": title,
+ "messages": messages
+ }
+ except Exception as e:
+ log.error("Error getting history: %s", e, exc_info=True)
+
+ return {
+ "thread_id": session_id,
+ "title": "New Conversation",
+ "messages": []
+ }
+
+ 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
+
+ async def close(self):
+ """
+ Close the checkpointer connection and cleanup resources.
+ """
+ async with self._init_lock:
+ if self._checkpointer_conn:
+ try:
+ await self._checkpointer_conn.close()
+ log.debug("Checkpointer connection closed for: %s", self.project_path)
+ except Exception as e:
+ log.warning("Error closing checkpointer connection: %s", e)
+ finally:
+ self._checkpointer_conn = None
+ self._checkpointer = None
+ self._graph = None
+ self._initialized = False
diff --git a/gns3server/agent/gns3_copilot/project_agent_manager.py b/gns3server/agent/gns3_copilot/project_agent_manager.py
new file mode 100644
index 000000000..cc71ea87b
--- /dev/null
+++ b/gns3server/agent/gns3_copilot/project_agent_manager.py
@@ -0,0 +1,121 @@
+"""
+Project Agent Manager
+
+Manages AgentService instances for GNS3 projects using a singleton pattern.
+Each project has its own AgentService with a dedicated SQLite checkpoint database.
+"""
+
+import asyncio
+import logging
+from typing import Dict, Optional
+
+from gns3server.agent.gns3_copilot.agent_service import AgentService
+
+log = logging.getLogger(__name__)
+
+
+class ProjectAgentManager:
+ """
+ Singleton manager for project-level Agent services.
+
+ Manages the lifecycle of AgentService instances, ensuring that each project
+ has exactly one AgentService instance. Handles cleanup of resources when
+ projects are closed.
+ """
+
+ _instance: Optional["ProjectAgentManager"] = None
+ _lock = asyncio.Lock()
+
+ def __new__(cls):
+ if cls._instance is None:
+ cls._instance = super().__new__(cls)
+ cls._instance._agents: Dict[str, AgentService] = {}
+ cls._instance._lock = asyncio.Lock()
+ return cls._instance
+
+ async def get_agent(self, project_id: str, project_path: str) -> AgentService:
+ """
+ Get or create an AgentService for a project.
+
+ Args:
+ project_id: GNS3 project ID
+ project_path: Path to the GNS3 project directory
+
+ Returns:
+ AgentService instance for the project
+ """
+ async with self._lock:
+ if project_id not in self._agents:
+ log.info("Creating new AgentService for project: %s at %s", project_id, project_path)
+ self._agents[project_id] = AgentService(project_path)
+ return self._agents[project_id]
+
+ async def remove_agent(self, project_id: str):
+ """
+ Remove and cleanup an AgentService for a project.
+
+ Should be called when a project is closed to free resources.
+
+ Args:
+ project_id: GNS3 project ID
+ """
+ async with self._lock:
+ if project_id in self._agents:
+ log.info("Removing AgentService for project: %s", project_id)
+ agent = self._agents.pop(project_id)
+ await agent.close()
+
+ async def close_all(self):
+ """
+ Close all AgentService instances and cleanup resources.
+
+ Should be called on server shutdown.
+ """
+ async with self._lock:
+ log.info("Closing all AgentService instances (%d projects)", len(self._agents))
+ for project_id, agent in self._agents.items():
+ log.debug("Closing AgentService for project: %s", project_id)
+ await agent.close()
+ self._agents.clear()
+
+ def has_agent(self, project_id: str) -> bool:
+ """
+ Check if an AgentService exists for a project.
+
+ Args:
+ project_id: GNS3 project ID
+
+ Returns:
+ True if AgentService exists, False otherwise
+ """
+ return project_id in self._agents
+
+ @property
+ def active_projects(self) -> list[str]:
+ """
+ Get list of project IDs with active AgentService instances.
+
+ Returns:
+ List of project IDs
+ """
+ return list(self._agents.keys())
+
+
+# Global singleton instance
+_project_agent_manager: Optional[ProjectAgentManager] = None
+_manager_lock = asyncio.Lock()
+
+
+async def get_project_agent_manager() -> ProjectAgentManager:
+ """
+ Get the global ProjectAgentManager singleton instance.
+
+ Returns:
+ ProjectAgentManager instance
+ """
+ global _project_agent_manager
+ async with _manager_lock:
+ if _project_agent_manager is None:
+ _project_agent_manager = ProjectAgentManager()
+ log.info("ProjectAgentManager singleton created")
+ return _project_agent_manager
diff --git a/gns3server/api/routes/controller/__init__.py b/gns3server/api/routes/controller/__init__.py
index 922d6ba25..26aabb4c7 100644
--- a/gns3server/api/routes/controller/__init__.py
+++ b/gns3server/api/routes/controller/__init__.py
@@ -16,6 +16,7 @@
from fastapi import APIRouter, Depends
+from . import chat
from . import controller
from . import appliances
from . import computes
@@ -152,3 +153,9 @@ router.include_router(
prefix="/access",
tags=["LLM Model Configurations"]
)
+
+router.include_router(
+ chat.router,
+ prefix="/chat",
+ tags=["Chat"]
+)
diff --git a/gns3server/api/routes/controller/chat.py b/gns3server/api/routes/controller/chat.py
new file mode 100644
index 000000000..856162b75
--- /dev/null
+++ b/gns3server/api/routes/controller/chat.py
@@ -0,0 +1,223 @@
+#
+# Copyright (C) 2025 GNS3 Technologies Inc.
+#
+# This program 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 .
+
+"""
+API routes for GNS3 Copilot Chat integration.
+"""
+
+import json
+import logging
+import uuid
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi.responses import StreamingResponse
+from uuid import UUID
+
+from gns3server import schemas
+from gns3server.controller import Controller
+from gns3server.controller.controller_error import ControllerNotFoundError
+from gns3server.agent.gns3_copilot.project_agent_manager import get_project_agent_manager
+
+from .dependencies.authentication import get_current_active_user
+
+log = logging.getLogger(__name__)
+
+responses = {404: {"model": schemas.ErrorMessage, "description": "Resource not found"}}
+
+router = APIRouter(responses=responses)
+
+
+async def dep_project(project_id: UUID):
+ """
+ Dependency to retrieve a project.
+ """
+ controller = Controller.instance()
+ project = controller.get_project(str(project_id))
+ if not project:
+ raise ControllerNotFoundError(f"Project '{project_id}' not found")
+ return project
+
+
+@router.post(
+ "/stream",
+ response_model=None,
+ summary="Stream chat responses from GNS3 Copilot",
+ description="Send a message to GNS3 Copilot and stream the response via Server-Sent Events (SSE)."
+)
+async def stream_chat(
+ request: schemas.ChatRequest,
+ http_request: Request,
+ current_user: schemas.User = Depends(get_current_active_user),
+) -> StreamingResponse:
+ """
+ Stream chat endpoint for GNS3 Copilot.
+
+ This endpoint uses Server-Sent Events (SSE) to stream responses from
+ the AI agent. Each message is a JSON object with a `type` field indicating
+ the message kind (content, tool_call, tool_start, tool_end, error, done, heartbeat).
+ """
+
+ # Validate project exists and get path
+ try:
+ controller = Controller.instance()
+ project = controller.get_project(str(request.project_id))
+ if not project:
+ raise ControllerNotFoundError(f"Project '{request.project_id}' not found")
+ project_path = project.path
+ except ControllerNotFoundError:
+ raise
+ except Exception as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid project: {e}"
+ )
+
+ # Get user authentication info
+ user_id = str(current_user.user_id)
+
+ # Get JWT token from Authorization header
+ auth_header = http_request.headers.get("Authorization", "")
+ jwt_token = auth_header.replace("Bearer ", "") if auth_header else None
+
+ # Get or create AgentService for this project
+ agent_manager = await get_project_agent_manager()
+ agent_service = await agent_manager.get_agent(str(request.project_id), project_path)
+
+ # Generate session_id if not provided
+ session_id = request.session_id or str(uuid.uuid4())
+
+ async def generate():
+ """Generator for SSE streaming."""
+ try:
+ async for chunk in agent_service.stream_chat(
+ message=request.message,
+ session_id=session_id,
+ project_id=str(request.project_id),
+ user_id=user_id,
+ jwt_token=jwt_token,
+ mode=request.mode
+ ):
+ try:
+ # Validate and serialize chunk
+ validated = schemas.ChatResponse(**chunk)
+ yield f"data: {json.dumps(validated.model_dump(exclude_none=True), ensure_ascii=False)}\n\n"
+ except Exception as e:
+ log.warning("Error serializing chunk: %s", e)
+ # Skip invalid chunks but continue streaming
+ continue
+
+ # Final done message
+ yield f"data: {json.dumps({'type': 'done', 'session_id': session_id})}\n\n"
+
+ except Exception as e:
+ log.error("Error in stream_chat: %s", e, exc_info=True)
+ yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n"
+
+ return StreamingResponse(
+ generate(),
+ media_type="text/event-stream",
+ headers={
+ "Cache-Control": "no-cache",
+ "X-Accel-Buffering": "no",
+ "Connection": "keep-alive",
+ }
+ )
+
+
+@router.get(
+ "/history/{session_id}",
+ response_model=schemas.ConversationHistory,
+ summary="Get conversation history",
+ description="Retrieve the conversation history for a specific session/thread."
+)
+async def get_history(
+ session_id: str,
+ project_id: UUID,
+ limit: int = 100,
+ current_user: schemas.User = Depends(get_current_active_user),
+) -> schemas.ConversationHistory:
+ """
+ Get conversation history for a session.
+ """
+
+ # Validate project exists
+ try:
+ controller = Controller.instance()
+ project = controller.get_project(str(project_id))
+ if not project:
+ raise ControllerNotFoundError(f"Project '{project_id}' not found")
+ project_path = project.path
+ except ControllerNotFoundError:
+ raise
+ except Exception as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid project: {e}"
+ )
+
+ # Get AgentService for this project
+ agent_manager = await get_project_agent_manager()
+ agent_service = await agent_manager.get_agent(str(project_id), project_path)
+
+ # Get history
+ history = await agent_service.get_history(session_id, limit)
+
+ return schemas.ConversationHistory(**history)
+
+
+@router.get(
+ "/sessions",
+ response_model=List[schemas.ChatSession],
+ summary="List chat sessions",
+ description="List all chat sessions for a project (not yet implemented)."
+)
+async def list_sessions(
+ project_id: UUID,
+ current_user: schemas.User = Depends(get_current_active_user),
+) -> list[schemas.ChatSession]:
+ """
+ List chat sessions for a project.
+
+ Note: This endpoint is a placeholder. Full session listing functionality
+ requires checkpoint metadata inspection which is not yet implemented.
+ """
+ # TODO: Implement session listing from checkpoint metadata
+ return []
+
+
+@router.delete(
+ "/sessions/{session_id}",
+ status_code=status.HTTP_204_NO_CONTENT,
+ summary="Delete a chat session",
+ description="Delete a specific chat session (not yet implemented)."
+)
+async def delete_session(
+ session_id: str,
+ project_id: UUID,
+ current_user: schemas.User = Depends(get_current_active_user),
+):
+ """
+ Delete a chat session.
+
+ Note: This endpoint is a placeholder. Full session deletion functionality
+ requires checkpoint manipulation which is not yet implemented.
+ """
+ # TODO: Implement session deletion from checkpoint
+ raise HTTPException(
+ status_code=status.HTTP_501_NOT_IMPLEMENTED,
+ detail="Session deletion not yet implemented"
+ )
diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py
index e73602d54..ffc77899e 100644
--- a/gns3server/controller/project.py
+++ b/gns3server/controller/project.py
@@ -849,6 +849,9 @@ class Project:
if not ignore_notification:
self.emit_controller_notification("project.closed", self.asdict())
+ # Cleanup GNS3 Copilot AgentService for this project
+ await self._cleanup_copilot_agent()
+
self.reset()
self._closing = False
@@ -885,6 +888,23 @@ class Project:
except OSError as e:
log.warning(f"Could not delete unused pictures: {e}")
+ async def _cleanup_copilot_agent(self):
+ """
+ Cleanup GNS3 Copilot AgentService for this project.
+
+ This should be called when the project is closed to free resources.
+ """
+ try:
+ from gns3server.agent.gns3_copilot.project_agent_manager import get_project_agent_manager
+
+ agent_manager = await get_project_agent_manager()
+ if agent_manager.has_agent(self._id):
+ log.info(f"Cleaning up AgentService for project '{self.name}' ({self._id})")
+ await agent_manager.remove_agent(self._id)
+ except Exception as e:
+ # Don't fail project close if agent cleanup fails
+ log.warning(f"Failed to cleanup AgentService for project '{self.name}': {e}")
+
async def delete(self):
if self._status != "opened":
diff --git a/gns3server/schemas/controller/chat.py b/gns3server/schemas/controller/chat.py
new file mode 100644
index 000000000..e94ee850e
--- /dev/null
+++ b/gns3server/schemas/controller/chat.py
@@ -0,0 +1,96 @@
+#
+# Copyright (C) 2025 GNS3 Technologies Inc.
+#
+# This program 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 .
+
+"""
+Chat API schemas for GNS3 Copilot integration.
+"""
+
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any, Literal
+
+
+class OpenAIToolCall(BaseModel):
+ """Tool call information (OpenAI compatible format)."""
+
+ id: str = Field(..., description="Tool call ID")
+ type: Literal["function"] = Field(default="function", description="Tool call type")
+ function: Dict[str, Any] = Field(..., description="Function name and arguments")
+
+
+class ChatRequest(BaseModel):
+ """Chat request model."""
+
+ message: str = Field(..., description="User message content")
+ session_id: Optional[str] = Field(None, description="Session ID (auto-generated if not provided)")
+ project_id: str = Field(..., description="GNS3 project ID")
+ stream: bool = Field(default=True, description="Enable streaming response")
+ temperature: Optional[float] = Field(None, description="LLM temperature parameter")
+ mode: Literal["text"] = Field(default="text", description="Interaction mode")
+
+
+class ChatResponse(BaseModel):
+ """Chat streaming response model."""
+
+ type: Literal[
+ "content", # AI text content
+ "tool_call", # Tool call request
+ "tool_start", # Tool execution started
+ "tool_end", # Tool execution completed
+ "error", # Error message
+ "done", # Stream ended
+ "heartbeat" # Keep-alive signal
+ ] = Field(..., description="Response message type")
+ content: Optional[str] = Field(None, description="Text content (for type=content)")
+ message_id: Optional[str] = Field(None, description="Message ID")
+ tool_call: Optional[OpenAIToolCall] = Field(None, description="Tool call (for type=tool_call)")
+ tool_name: Optional[str] = Field(None, description="Tool name (for type=tool_start/end)")
+ tool_output: Optional[str] = Field(None, description="Tool output (for type=tool_end)")
+ error: Optional[str] = Field(None, description="Error message (for type=error)")
+ session_id: Optional[str] = Field(None, description="Session ID (for type=heartbeat/done)")
+
+
+class OpenAIMessage(BaseModel):
+ """Message model for conversation history."""
+
+ id: str = Field(..., description="Message ID")
+ role: Literal["user", "assistant", "system", "tool"] = Field(..., description="Message role")
+ content: str = Field(..., description="Message content")
+ name: Optional[str] = Field(None, description="Tool message name")
+ tool_call_id: Optional[str] = Field(None, description="Associated tool call ID (for tool messages)")
+ tool_calls: Optional[List[OpenAIToolCall]] = Field(None, description="Tool calls (for assistant messages)")
+ metadata: Dict[str, Any] = Field(default_factory=dict, description="Message metadata")
+ created_at: str = Field(..., description="Message timestamp (ISO 8601)")
+
+
+class ConversationHistory(BaseModel):
+ """Conversation history model."""
+
+ thread_id: str = Field(..., description="Thread/session ID")
+ title: str = Field(..., description="Conversation title")
+ messages: List[OpenAIMessage] = Field(default_factory=list, description="Conversation messages")
+ created_at: Optional[str] = Field(None, description="Creation timestamp (ISO 8601)")
+ updated_at: Optional[str] = Field(None, description="Last update timestamp (ISO 8601)")
+ llm_calls: int = Field(default=0, description="Total LLM calls in this conversation")
+
+
+class ChatSession(BaseModel):
+ """Chat session model."""
+
+ session_id: str = Field(..., description="Session ID")
+ title: str = Field(..., description="Session title")
+ project_id: Optional[str] = Field(None, description="Associated GNS3 project ID")
+ created_at: Optional[str] = Field(None, description="Creation timestamp (ISO 8601)")
+ updated_at: Optional[str] = Field(None, description="Last update timestamp (ISO 8601)")