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.