get_queries_using_primary_index

Retrieve queries using primary indexes from Couchbase's system catalog to identify optimization opportunities by analyzing completed requests.

Instructions

Get queries that use a primary index from the system:completed_requests catalog.

Args:
    limit: Number of queries to return (default: 10)

Returns:
    List of queries that use primary indexes, ordered by result count

Input Schema

TableJSON Schema

Name	Required	Description	Default
`limit`	No

Implementation Reference

src/tools/query.py:272-299 (handler)

The main handler function that executes the tool logic. It runs a SQL++ query on system:completed_requests to find queries using primary scans, ordered by resultCount, with a custom empty message.

def get_queries_using_primary_index(
    ctx: Context, limit: int = 10
) -> list[dict[str, Any]]:
    """Get queries that use a primary index from the system:completed_requests catalog.

    Args:
        limit: Number of queries to return (default: 10)

    Returns:
        List of queries that use primary indexes, ordered by result count
    """
    query = """
    SELECT *
    FROM system:completed_requests
    WHERE phaseCounts.`primaryScan` IS NOT MISSING
        AND UPPER(statement) NOT LIKE '% SYSTEM:%'
    ORDER BY resultCount DESC
    LIMIT $limit
    """

    return _run_query_tool_with_empty_message(
        ctx,
        query,
        limit=limit,
        empty_message=(
            "No queries using the primary index were found in system:completed_requests."
        ),
    )

src/mcp_server.py:173-180 (registration)

The tool is registered in the FastMCP server by iterating over ALL_TOOLS from src/tools and calling mcp.add_tool(tool) for each.

mcp = FastMCP(MCP_SERVER_NAME, lifespan=app_lifespan, **config)

# Register all tools
for tool in ALL_TOOLS:
    mcp.add_tool(tool)

# Run the server
mcp.run(transport=sdk_transport)  # type: ignore

src/tools/__init__.py:57-64 (registration)

The tool function is included in the ALL_TOOLS list, which is imported and used for registration in mcp_server.py.

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

src/tools/query.py:108-127 (helper)

Helper function used by get_queries_using_primary_index (and other similar tools) to run the cluster query and handle empty results with a custom 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]

Couchbase MCP Server