Mnemosyne MCP

Instructions

Implementation Reference

• src/neem/mcp/server/standalone_server.py:263-326 (handler)

  Handler function for the 'create_session' MCP tool. Creates a new session using the session manager, authenticates the user, and returns session details as JSON.
  @mcp_server.tool() async def create_session( ctx: Context, client_name: str = "Unknown Client" ) -> str: """ Create new MCP session for the authenticated user. Args: client_name: Name/type of the MCP client Returns: JSON string with session information """ try: # Extract user ID from auth token auth_token = extract_auth_token(ctx) user_id = auth_token # In our system, token is user_id for development logger.info(f"Creating MCP session for user {user_id}") # Create session with client info user_agent = 'Unknown' try: if ctx and ctx.request_context and ctx.request_context.request and hasattr(ctx.request_context.request, 'headers'): headers = ctx.request_context.request.headers if headers: user_agent = getattr(headers, 'user-agent', 'Unknown') except (AttributeError, TypeError): pass # Use default 'Unknown' client_info = { "client_name": client_name, "user_agent": user_agent } session_id = await session_manager.create_session(user_id, client_info) # Get session data for response session_data = await session_manager.get_user_session(user_id, session_id) result = { "success": True, "session_id": session_id, "user_id": user_id, "accessible_graphs": session_data.get("accessible_graphs", []), "created_at": session_data.get("created_at"), "instructions": { "next_step": "Include session ID in X-Session-ID header for subsequent requests", "session_ttl": "1 hour" } } return json.dumps(result, indent=2) except Exception as e: logger.error(f"Session creation failed: {e}") error_result = { "success": False, "error": str(e), "error_type": "SESSION_CREATION_FAILED" } return json.dumps(error_result, indent=2)
• src/neem/mcp/server/standalone_server.py:263-263 (registration)

  Registration of the 'create_session' tool via FastMCP @tool() decorator.
  @mcp_server.tool()
• src/neem/mcp/session.py:64-116 (helper)

  Supporting method in MCPSessionManager class that actually creates and stores the session data in memory with TTL.
  async def create_session( self, user_id: str, client_info: Optional[Dict[str, Any]] = None ) -> str: """ Create new MCP session for user. Args: user_id: User identifier client_info: Optional client information Returns: Session ID Raises: MCPSessionError: If session creation fails """ try: # Generate unique session ID session_id = str(uuid.uuid4()) session_key = f"{self.session_prefix}:{user_id}:{session_id}" # Note: In standalone mode, accessible_graphs will be populated by API queries # when needed, rather than storing them in session data accessible_graphs = [] # Create session data session_data = { "user_id": user_id, "session_id": session_id, "created_at": datetime.utcnow().isoformat(), "last_activity": datetime.utcnow().isoformat(), "accessible_graphs": accessible_graphs, "client_info": client_info or {}, "status": "active" } expires_at = datetime.utcnow() + timedelta(seconds=self.session_ttl) async with self._lock: self._sessions[session_key] = session_data self._expirations[session_key] = expires_at logger.info(f"Created MCP session {session_id} for user {user_id}") return session_id except Exception as e: logger.error(f"Failed to create MCP session for user {user_id}: {e}") raise MCPSessionError( f"Session creation failed: {str(e)}", context={"user_id": user_id} )
• src/neem/mcp/server/standalone_server.py:138-142 (helper)

  Factory function that creates the MCPSessionManager instance used by the tool.
  def create_session_manager() -> MCPSessionManager: """Create in-memory session manager.""" session_manager = MCPSessionManager() logger.info("✅ In-memory session manager created") return session_manager