Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/query.py:198-234 (handler)

  The core handler function that implements the tool logic by executing a SQL++ query on system:completed_requests to identify queries with the largest average response sizes, formatting sizes in bytes, KB, and MB, and handling empty results gracefully.
  def get_queries_with_largest_response_sizes( ctx: Context, limit: int = 10 ) -> list[dict[str, Any]]: """Get queries with the largest response sizes from the system:completed_requests catalog. Args: limit: Number of queries to return (default: 10) Returns: List of queries with their average result size in bytes, KB, and MB """ query = """ SELECT statement, avgResultSize AS avgResultSizeBytes, (avgResultSize / 1000) AS avgResultSizeKB, (avgResultSize / 1000000) AS avgResultSizeMB, 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 avgResultSize = AVG(resultSize) ORDER BY avgResultSize DESC LIMIT $limit """ return _run_query_tool_with_empty_message( ctx, query, limit=limit, empty_message=( "No completed queries were available to calculate response sizes." ), )
• src/mcp_server.py:175-177 (registration)

  The tool is registered to the FastMCP server instance by iterating over ALL_TOOLS (which includes this tool) and calling add_tool().
  # Register all tools for tool in ALL_TOOLS: mcp.add_tool(tool)
• src/tools/__init__.py:58-61 (registration)

  The tool handler is imported from query.py and explicitly included in the ALL_TOOLS list, which is used by mcp_server.py for registration.
  get_queries_not_using_covering_index, get_queries_using_primary_index, get_queries_with_large_result_count, get_queries_with_largest_response_sizes,
• src/tools/query.py:108-127 (helper)

  Supporting helper utility invoked by the handler to run the cluster query and provide a standardized empty results response.
  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]