Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/query.py:20-36 (handler)

  The handler function implementing the get_schema_for_collection tool. It runs an 'INFER' SQL++ query on the specified bucket.scope.collection to retrieve and return the schema information.

  def get_schema_for_collection(
      ctx: Context, bucket_name: str, scope_name: str, collection_name: str
  ) -> dict[str, Any]:
      """Get the schema for a collection in the specified scope.
      Returns a dictionary with the collection name and the schema returned by running INFER query on the Couchbase collection.
      """
      schema = {"collection_name": collection_name, "schema": []}
      try:
          query = f"INFER `{collection_name}`"
          result = run_sql_plus_plus_query(ctx, bucket_name, scope_name, query)
          # Result is a list of list of schemas. We convert it to a list of schemas.
          if result:
              schema["schema"] = result[0]
      except Exception as e:
          logger.error(f"Error getting schema: {e}")
          raise
      return schema

• src/mcp_server.py:176-178 (registration)

  Registers the get_schema_for_collection tool (along with others from ALL_TOOLS) with the FastMCP server instance.

  for tool in ALL_TOOLS:
      mcp.add_tool(tool)

• src/tools/__init__.py:52-52 (registration)

  Includes get_schema_for_collection in the ALL_TOOLS list used for server registration.

  get_schema_for_collection,

• src/tools/__init__.py:18-28 (registration)

  Imports the get_schema_for_collection handler from query.py for exposure via __init__.

  from .query import (
      get_longest_running_queries,
      get_most_frequent_queries,
      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_schema_for_collection,
      run_sql_plus_plus_query,
  )