Couchbase MCP Server

Instructions

Implementation Reference

• src/tools/kv.py:19-37 (handler)

  The core handler function for the 'get_document_by_id' tool. Retrieves a document from Couchbase using the cluster connection, bucket, scope, collection, and document ID. Returns the document content as dict or raises exception.

  def get_document_by_id(
      ctx: Context,
      bucket_name: str,
      scope_name: str,
      collection_name: str,
      document_id: str,
  ) -> dict[str, Any]:
      """Get a document by its ID from the specified scope and collection.
      If the document is not found, it will raise an exception."""
  
      cluster = get_cluster_connection(ctx)
      bucket = connect_to_bucket(cluster, bucket_name)
      try:
          collection = bucket.scope(scope_name).collection(collection_name)
          result = collection.get(document_id)
          return result.content_as[dict]
      except Exception as e:
          logger.error(f"Error getting document {document_id}: {e}")
          raise

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

  Imports the get_document_by_id handler from the kv module into the tools package __init__, exposing it for use in ALL_TOOLS and server registration.

  from .kv import (
      delete_document_by_id,
      get_document_by_id,
      upsert_document_by_id,
  )

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

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

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