feat(copilot): refactor JWT token handling and improve metadata tracking

- Move JWT token from state to configurable context for better security and request isolation
- Add user_id parameter to agent service for enhanced metadata tracking
- Update checkpoint directory name from .gns3-copilot to gns3-copilot
- Implement context-aware JWT token management using ContextVar
- Improve tool node to extract JWT token from config instead of state
This commit is contained in:
YueGuobin 2026-03-04 13:36:16 +08:00
parent 5e9aac7514
commit b05e6a71b4
5 changed files with 43 additions and 7 deletions

View File

@ -214,12 +214,10 @@ def llm_call(state: dict, config: RunnableConfig | None = None):
llm_config=llm_config
)
# 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,
}
@ -305,9 +303,18 @@ def generate_title(state: MessagesState, config: RunnableConfig | None = None) -
# Define tool node
def tool_node(state: dict):
def tool_node(state: dict, config: RunnableConfig | None = None):
"""Performs the tool call"""
# Extract jwt_token from config for tools to use
configurable = config.get("configurable", {}) if config else {}
jwt_token = configurable.get("jwt_token")
# Set jwt_token for connector_factory to access
if jwt_token:
from gns3server.agent.gns3_copilot.gns3_client import set_current_jwt_token
set_current_jwt_token(jwt_token)
result = []
for tool_call in state["messages"][-1].tool_calls:
tool = tools_by_name[tool_call["name"]]

View File

@ -46,7 +46,7 @@ class AgentService:
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")
checkpoint_dir = os.path.join(self.project_path, "gns3-copilot")
os.makedirs(checkpoint_dir, exist_ok=True)
return checkpoint_dir
@ -103,6 +103,7 @@ class AgentService:
message: str,
session_id: str,
project_id: Optional[str] = None,
user_id: Optional[str] = None,
jwt_token: Optional[str] = None,
mode: str = "text",
llm_config: Optional[Dict[str, Any]] = None
@ -114,6 +115,7 @@ class AgentService:
message: User message
session_id: Session/thread ID for conversation continuity
project_id: GNS3 project ID (optional, for context)
user_id: User ID for metadata tracking
jwt_token: JWT token for API authentication (optional)
mode: Interaction mode (default: "text")
llm_config: LLM configuration dict (provider, model, api_key, etc.)
@ -121,12 +123,16 @@ class AgentService:
Yields:
Dict containing SSE-compatible response chunks
"""
# Build config with LLM configuration
# Build config with LLM configuration and metadata
config = {
"configurable": {
"thread_id": session_id,
"jwt_token": jwt_token,
"llm_config": llm_config,
},
"metadata": {
"user_id": user_id,
"project_id": project_id,
}
}

View File

@ -25,6 +25,8 @@ from .connector_factory import (
get_gns3_connector_with_llm_config,
get_gns3_server_host,
get_llm_config,
set_current_jwt_token,
get_current_jwt_token,
)
from .custom_gns3fy import (
CONSOLE_TYPES,
@ -62,4 +64,6 @@ __all__ = [
"get_gns3_connector_with_llm_config",
"get_gns3_server_host",
"get_llm_config",
"set_current_jwt_token",
"get_current_jwt_token",
]

View File

@ -28,6 +28,21 @@ Authentication:
import logging
from typing import Optional
from contextvars import ContextVar
# Context variable for request-scoped jwt_token
# Automatically cleaned up when request context ends
_jwt_token_context: ContextVar[Optional[str]] = ContextVar("_jwt_token_context", default=None)
def set_current_jwt_token(token: str) -> None:
"""Set the JWT token for the current request context."""
_jwt_token_context.set(token)
def get_current_jwt_token() -> Optional[str]:
"""Get the JWT token for the current request context."""
return _jwt_token_context.get()
from uuid import UUID
from gns3server.agent.gns3_copilot.gns3_client.custom_gns3fy import Gns3Connector
@ -136,9 +151,12 @@ def get_gns3_connector(jwt_token: str, url: Optional[str] = None) -> Optional[Gn
"""
try:
# Validate JWT token
# If not provided, try to get from current request context
if not jwt_token:
logger.error("JWT token parameter is required")
return None
jwt_token = get_current_jwt_token()
if not jwt_token:
logger.error("JWT token is required")
return None
# Resolve URL with fallback strategy
if url is None:

View File

@ -117,6 +117,7 @@ async def stream_chat(
message=request.message,
session_id=session_id,
project_id=str(project.id),
user_id=user_id,
jwt_token=jwt_token,
mode=request.mode,
llm_config=llm_config