get_longest_running_queries
Identify and retrieve the longest-running queries from Couchbase's completed_requests catalog to analyze performance bottlenecks and optimize database efficiency.
Instructions
Get the N longest running queries from the system:completed_requests catalog.
Args:
limit: Number of queries to return (default: 10)
Returns:
List of queries with their average service time and countInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/query.py:129-160 (handler)The primary handler function implementing the 'get_longest_running_queries' tool. It runs a SQL++ query on system:completed_requests to fetch the longest running queries grouped by statement, ordered by average service time.
def get_longest_running_queries(ctx: Context, limit: int = 10) -> list[dict[str, Any]]: """Get the N longest running queries from the system:completed_requests catalog. Args: limit: Number of queries to return (default: 10) Returns: List of queries with their average service time and count """ query = """ SELECT statement, DURATION_TO_STR(avgServiceTime) AS avgServiceTime, COUNT(1) AS queries FROM system:completed_requests WHERE UPPER(statement) NOT LIKE 'INFER %' AND UPPER(statement) NOT LIKE 'CREATE INDEX%' AND UPPER(statement) NOT LIKE 'CREATE PRIMARY INDEX%' AND UPPER(statement) NOT LIKE '% SYSTEM:%' GROUP BY statement LETTING avgServiceTime = AVG(STR_TO_DURATION(serviceTime)) ORDER BY avgServiceTime DESC LIMIT $limit """ return _run_query_tool_with_empty_message( ctx, query, limit=limit, empty_message=( "No completed queries were available to calculate longest running queries." ), ) - src/tools/query.py:108-127 (helper)Helper function called by get_longest_running_queries to execute the cluster query and handle empty results with a standard message.
def _run_query_tool_with_empty_message( ctx: Context, query: str, *, limit: int, empty_message: str, extra_payload: dict[str, Any] | None = None, **query_kwargs: Any, ) -> list[dict[str, Any]]: """Execute a cluster query with a consistent empty-result response.""" results = run_cluster_query(ctx, query, limit=limit, **query_kwargs) if results: return results payload: dict[str, Any] = {"message": empty_message, "results": []} if extra_payload: payload.update(extra_payload) return [payload] - src/mcp_server.py:175-178 (registration)Registration loop in the MCP server where all tools, including get_longest_running_queries (imported via ALL_TOOLS), are added to the FastMCP server instance.
# Register all tools for tool in ALL_TOOLS: mcp.add_tool(tool) - src/tools/__init__.py:42-64 (registration)Definition of ALL_TOOLS list in tools/__init__.py which includes get_longest_running_queries and is used for bulk registration in mcp_server.py.
ALL_TOOLS = [ get_buckets_in_cluster, get_server_configuration_status, test_cluster_connection, get_scopes_and_collections_in_bucket, get_collections_in_scope, get_scopes_in_bucket, get_document_by_id, upsert_document_by_id, delete_document_by_id, get_schema_for_collection, run_sql_plus_plus_query, get_index_advisor_recommendations, list_indexes, get_cluster_health_and_services, get_queries_not_selective, get_queries_not_using_covering_index, get_queries_using_primary_index, get_queries_with_large_result_count, get_queries_with_largest_response_sizes, get_longest_running_queries, get_most_frequent_queries, ]