Couchbase MCP Server

Instructions

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]