5月28日 06:55
How to implement session management and context maintenance in MCP?
Session management and context maintenance in MCP are critical for ensuring conversation continuity and state consistency. Here are detailed implementation methods:
Session Management Basics
MCP sessions include session ID, context data, state information, etc.:
{ "session_id": "unique-session-id", "context": {}, "state": "active", "created_at": "2024-01-01T00:00:00Z", "last_activity": "2024-01-01T00:00:00Z" }
1. Session Creation and Management
import uuid from datetime import datetime from typing import Dict, Optional class SessionManager: def __init__(self): self.sessions = {} self.session_timeout = 3600 # 1 hour def create_session(self, initial_context: dict = None) -> str: """Create new session""" session_id = str(uuid.uuid4()) self.sessions[session_id] = { "session_id": session_id, "context": initial_context or {}, "state": "active", "created_at": datetime.now(), "last_activity": datetime.now(), "message_history": [] } return session_id def get_session(self, session_id: str) -> Optional[dict]: """Get session""" if session_id not in self.sessions: return None session = self.sessions[session_id] # Check if session is expired if self._is_session_expired(session): self.close_session(session_id) return None return session def update_session(self, session_id: str, updates: dict): """Update session""" session = self.get_session(session_id) if not session: raise ValueError(f"Session {session_id} does not exist or has expired") session.update(updates) session["last_activity"] = datetime.now() def close_session(self, session_id: str): """Close session""" if session_id in self.sessions: self.sessions[session_id]["state"] = "closed" del self.sessions[session_id] def _is_session_expired(self, session: dict) -> bool: """Check if session is expired""" elapsed = (datetime.now() - session["last_activity"]).total_seconds() return elapsed > self.session_timeout
2. Context Management
class ContextManager: def __init__(self, session_manager: SessionManager): self.session_manager = session_manager def set_context( self, session_id: str, key: str, value: any ): """Set context value""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") session["context"][key] = value self.session_manager.update_session(session_id, { "context": session["context"] }) def get_context( self, session_id: str, key: str, default: any = None ) -> any: """Get context value""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") return session["context"].get(key, default) def update_context( self, session_id: str, updates: dict ): """Batch update context""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") session["context"].update(updates) self.session_manager.update_session(session_id, { "context": session["context"] }) def clear_context(self, session_id: str): """Clear context""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") session["context"] = {} self.session_manager.update_session(session_id, { "context": session["context"] })
3. Message History Management
from typing import List class MessageHistoryManager: def __init__(self, session_manager: SessionManager): self.session_manager = session_manager self.max_history = 100 # Maximum 100 messages def add_message( self, session_id: str, role: str, content: str, metadata: dict = None ): """Add message to history""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") message = { "role": role, # "user", "assistant", "system" "content": content, "timestamp": datetime.now().isoformat(), "metadata": metadata or {} } session["message_history"].append(message) # Limit history size if len(session["message_history"]) > self.max_history: session["message_history"] = \ session["message_history"][-self.max_history:] self.session_manager.update_session(session_id, { "message_history": session["message_history"] }) def get_history( self, session_id: str, limit: int = None ) -> List[dict]: """Get message history""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") history = session["message_history"] if limit: return history[-limit:] return history def get_conversation_summary( self, session_id: str ) -> dict: """Get conversation summary""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") history = session["message_history"] return { "total_messages": len(history), "user_messages": len([ m for m in history if m["role"] == "user" ]), "assistant_messages": len([ m for m in history if m["role"] == "assistant" ]), "first_message": history[0] if history else None, "last_message": history[-1] if history else None }
4. State Machine Management
from enum import Enum class SessionState(Enum): ACTIVE = "active" IDLE = "idle" SUSPENDED = "suspended" CLOSED = "closed" class StateMachine: def __init__(self, session_manager: SessionManager): self.session_manager = session_manager self.transitions = { SessionState.ACTIVE: [SessionState.IDLE, SessionState.SUSPENDED, SessionState.CLOSED], SessionState.IDLE: [SessionState.ACTIVE, SessionState.CLOSED], SessionState.SUSPENDED: [SessionState.ACTIVE, SessionState.CLOSED], SessionState.CLOSED: [] } def transition_state( self, session_id: str, new_state: SessionState ): """Transition session state""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") current_state = SessionState(session["state"]) # Validate state transition if new_state not in self.transitions.get(current_state, []): raise ValueError( f"Invalid state transition: {current_state} -> {new_state}" ) # Execute state transition self.session_manager.update_session(session_id, { "state": new_state.value }) def get_state(self, session_id: str) -> SessionState: """Get current state""" session = self.session_manager.get_session(session_id) if not session: raise ValueError("Session does not exist") return SessionState(session["state"])
5. Session Persistence
import json import os from typing import Optional class SessionPersistence: def __init__(self, storage_path: str): self.storage_path = storage_path os.makedirs(storage_path, exist_ok=True) def save_session(self, session_id: str, session: dict): """Save session to disk""" file_path = os.path.join( self.storage_path, f"{session_id}.json" ) # Convert datetime objects to strings session_copy = session.copy() for key, value in session_copy.items(): if isinstance(value, datetime): session_copy[key] = value.isoformat() with open(file_path, 'w') as f: json.dump(session_copy, f, indent=2) def load_session(self, session_id: str) -> Optional[dict]: """Load session from disk""" file_path = os.path.join( self.storage_path, f"{session_id}.json" ) if not os.path.exists(file_path): return None with open(file_path, 'r') as f: session = json.load(f) # Convert strings to datetime objects for key in ["created_at", "last_activity"]: if key in session and isinstance(session[key], str): session[key] = datetime.fromisoformat(session[key]) return session def delete_session(self, session_id: str): """Delete session file""" file_path = os.path.join( self.storage_path, f"{session_id}.json" ) if os.path.exists(file_path): os.remove(file_path) def list_sessions(self) -> List[str]: """List all sessions""" sessions = [] for filename in os.listdir(self.storage_path): if filename.endswith('.json'): session_id = filename[:-5] # Remove .json extension sessions.append(session_id) return sessions
6. Session Monitoring and Analytics
from collections import defaultdict class SessionAnalytics: def __init__(self, session_manager: SessionManager): self.session_manager = session_manager self.metrics = defaultdict(int) def track_event(self, event_type: str, session_id: str = None): """Track event""" self.metrics[event_type] += 1 def get_metrics(self) -> dict: """Get metrics""" return dict(self.metrics) def get_session_statistics(self) -> dict: """Get session statistics""" sessions = self.session_manager.sessions return { "total_sessions": len(sessions), "active_sessions": len([ s for s in sessions.values() if s["state"] == "active" ]), "average_session_duration": self._calculate_avg_duration(sessions), "total_messages": sum( len(s["message_history"]) for s in sessions.values() ) } def _calculate_avg_duration(self, sessions: dict) -> float: """Calculate average session duration""" if not sessions: return 0.0 durations = [ (s["last_activity"] - s["created_at"]).total_seconds() for s in sessions.values() ] return sum(durations) / len(durations)
Best Practices:
- Session Isolation: Ensure complete context isolation between different sessions
- Timeout Management: Set reasonable session timeout and automatically clean up expired sessions
- Persistence Strategy: Implement persistence for important sessions to prevent data loss
- State Transitions: Use state machines to manage session state transitions
- Monitoring and Analytics: Monitor session metrics and analyze user behavior
- Resource Cleanup: Regularly clean up unused sessions and resources
Through comprehensive session management and context maintenance mechanisms, you can ensure conversation continuity and state consistency in MCP systems.