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 count

Input Schema

TableJSON Schema

Name	Required	Description	Default
`limit`	No

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,
]

Couchbase MCP Server