Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/kv.py:40-59 (handler)

  The handler function that executes the upsert_document_by_id tool logic, connecting to Couchbase and performing the upsert operation.

  def upsert_document_by_id(
      ctx: Context,
      bucket_name: str,
      scope_name: str,
      collection_name: str,
      document_id: str,
      document_content: dict[str, Any],
  ) -> bool:
      """Insert or update a document by its ID.
      Returns True on success, False on failure."""
      cluster = get_cluster_connection(ctx)
      bucket = connect_to_bucket(cluster, bucket_name)
      try:
          collection = bucket.scope(scope_name).collection(collection_name)
          collection.upsert(document_id, document_content)
          logger.info(f"Successfully upserted document {document_id}")
          return True
      except Exception as e:
          logger.error(f"Error upserting document {document_id}: {e}")
          return False

• src/mcp_server.py:175-177 (registration)

  Registers all tools, including upsert_document_by_id, with the FastMCP server instance.

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

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

  Imports and exposes the upsert_document_by_id function for inclusion in the ALL_TOOLS list used for registration.

  # Key-Value tools
  from .kv import (
      delete_document_by_id,
      get_document_by_id,
      upsert_document_by_id,
  )
  
  # Query tools
  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,
  )
  
  # Server tools
  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,
  )
  
  # List of all tools for easy registration
  ALL_TOOLS = [
      get_buckets_in_cluster,
      get_server_configuration_status,
      test_cluster_connection,
      get_scopes_and_collections_in_bucket,
      get_collections_in_scope,
      get_scopes_in_bucket,
      get_document_by_id,
      upsert_document_by_id,
      delete_document_by_id,
      get_schema_for_collection,
      run_sql_plus_plus_query,
      get_index_advisor_recommendations,
      list_indexes,
      get_cluster_health_and_services,
      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,
  ]