Mnemosyne MCP

Instructions

Implementation Reference

• src/neem/mcp/server/standalone_server.py:761-878 (handler)

  The primary handler function for the 'get_graph_info' tool, decorated with @mcp_server.tool() for registration. It authenticates via context, calls the API, handles errors, and returns a formatted JSON response using McpGraphInfoResponse.
  @mcp_server.tool() async def get_graph_info( graph_id: str, ctx: Context, include_stats: bool = True, include_schema_preview: bool = False ) -> str: """ Get comprehensive graph information. Args: graph_id: Graph to analyze include_stats: Include detailed statistics include_schema_preview: Include schema summary ctx: MCP context for user authentication Returns: JSON response with complete graph details """ operation_id = str(uuid.uuid4()) with logger.operation_context("get_graph_info", operation_id=operation_id, graph_id=graph_id): try: # Get session context auth_token, session_data = await get_session_context(ctx) user_id = auth_token # Validate graph access if session_data: accessible_graphs = session_data.get("accessible_graphs", []) if graph_id not in accessible_graphs: logger.warning( "Graph access denied for info request", extra_context={ "user_id": user_id, "graph_id": graph_id } ) error_response = McpErrorResponse( error_type="GRAPH_ACCESS_DENIED", error_message=f"Access denied for graph {graph_id}", help={ "graph_id": graph_id, "suggestion": "Use list_graphs to see accessible graphs" } ) return error_response.to_json() logger.info( "Getting graph information", extra_context={ "operation_id": operation_id, "user_id": user_id, "graph_id": graph_id, "include_stats": include_stats, "include_schema_preview": include_schema_preview } ) # Call API server params = {} if include_stats: params["include_stats"] = "true" if include_schema_preview: params["include_schema_preview"] = "true" api_response = await api_client._make_request( "GET", f"/graphs/{graph_id}", headers={"Authorization": f"Bearer {auth_token}"}, params=params ) logger.info( "Graph information retrieved successfully", extra_context={ "operation_id": operation_id, "graph_id": graph_id, "user_id": user_id } ) # Transform to enhanced MCP format mcp_response = McpGraphInfoResponse.from_api_response( api_response, operation_id=operation_id, graph_id=graph_id, user_id=user_id ) return mcp_response.to_json() except Exception as e: logger.error( "Graph info retrieval failed", extra_context={ "operation_id": operation_id, "graph_id": graph_id, "error": str(e), "error_type": type(e).__name__ } ) error_response = McpErrorResponse.from_exception( e, error_type="GRAPH_INFO_FAILED", help_info={ "operation_id": operation_id, "graph_id": graph_id, "suggestions": [ "Check that graph exists with list_graphs tool", "Verify graph access permissions", "Try with include_stats=false for basic info" ] } ) return error_response.to_json()
• src/neem/mcp/response_objects.py:605-656 (schema)

  Defines the output schema/response object McpGraphInfoResponse for the get_graph_info tool, including structured fields for graph info, statistics, and recommendations. Used by the handler to format API responses.
  class McpGraphInfoResponse(McpResponseObject): """Response for detailed graph information""" operation_summary: McpOperationSummary graph_info: Dict[str, Any] statistics: Dict[str, Any] schema_preview: Optional[Dict[str, Any]] recommendations: List[str] success: bool = True @classmethod def from_api_response(cls, api_response: Dict[str, Any], operation_id: str, graph_id: str, user_id: Optional[str] = None) -> 'McpGraphInfoResponse': """Transform API graph info response to MCP format""" import datetime graph_data = api_response.get("data", {}) # Generate recommendations based on graph state recommendations = [] triple_count = graph_data.get("triple_count", 0) if triple_count == 0: recommendations.append("Graph is empty - consider uploading RDF data") elif triple_count < 100: recommendations.append("Small graph - suitable for experimentation and learning") elif triple_count > 10000: recommendations.append("Large graph - consider using LIMIT in queries for better performance") return cls( operation_summary=McpOperationSummary( operation_type="get_graph_info", operation_id=operation_id, timestamp=datetime.datetime.now().isoformat(), user_id=user_id ), graph_info={ "id": graph_id, "name": graph_data.get("name", ""), "description": graph_data.get("description", ""), "created_at": graph_data.get("created_at"), "updated_at": graph_data.get("updated_at"), "is_persistent": graph_data.get("is_persistent", True) }, statistics={ "triple_count": triple_count, "size_mb": graph_data.get("size_mb", 0.0), "unique_subjects": graph_data.get("unique_subjects", 0), "unique_predicates": graph_data.get("unique_predicates", 0), "unique_objects": graph_data.get("unique_objects", 0) }, schema_preview=graph_data.get("schema_preview"), recommendations=recommendations )
• src/neem/mcp/server/standalone_server.py:1079-1080 (registration)

  Logs the list of available tools including 'get_graph_info' after server setup, confirming its registration.
  logger.info("Available tools: create_session, sparql_query, list_graphs, get_graph_schema, create_graph, delete_graph, get_graph_info, get_system_health, upload_file_to_graph")