Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/server.py:84-102 (handler)

  The main handler function that connects to the Couchbase cluster and bucket, retrieves all scopes using bucket.collections().get_all_scopes(), and maps each scope to its collections, returning a dictionary of scopes to lists of collection names.

  def get_scopes_and_collections_in_bucket(
      ctx: Context, bucket_name: str
  ) -> dict[str, list[str]]:
      """Get the names of all scopes and collections in the bucket.
      Returns a dictionary with scope names as keys and lists of collection names as values.
      """
      cluster = get_cluster_connection(ctx)
      bucket = connect_to_bucket(cluster, bucket_name)
      try:
          scopes_collections = {}
          collection_manager = bucket.collections()
          scopes = collection_manager.get_all_scopes()
          for scope in scopes:
              collection_names = [c.name for c in scope.collections]
              scopes_collections[scope.name] = collection_names
          return scopes_collections
      except Exception as e:
          logger.error(f"Error getting scopes and collections: {e}")
          raise

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

  Imports the get_scopes_and_collections_in_bucket handler from src/tools/server.py into the tools namespace.

  from .server import (
      get_buckets_in_cluster,
      get_cluster_health_and_services,
      get_collections_in_scope,
      get_scopes_and_collections_in_bucket,
      get_scopes_in_bucket,
      get_server_configuration_status,
      test_cluster_connection,
  )

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

  Includes the tool function in the ALL_TOOLS list, which is later used to register all tools with the MCP server.

  ALL_TOOLS = [
      get_buckets_in_cluster,
      get_server_configuration_status,
      test_cluster_connection,
      get_scopes_and_collections_in_bucket,

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

  Registers all tools from ALL_TOOLS, including get_scopes_and_collections_in_bucket, with the FastMCP server instance.

  for tool in ALL_TOOLS:
      mcp.add_tool(tool)