Couchbase MCP Server

Instructions

Implementation Reference

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

  The main handler function for the 'get_schema_for_collection' tool. It executes an INFER SQL++ query on the specified Couchbase collection to infer and return its schema.
  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:175-177 (registration)

  Registration of all tools (including get_schema_for_collection) with the FastMCP server instance using a loop over ALL_TOOLS.
  # Register all tools for tool in ALL_TOOLS: mcp.add_tool(tool)
• src/tools/__init__.py:35-50 (registration)

  The ALL_TOOLS list includes get_schema_for_collection, which is used by the MCP server to register all available tools.
  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, ]
• src/tools/query.py:39-90 (helper)

  Supporting helper function run_sql_plus_plus_query called by the handler to execute the INFER query, with read-only mode checks and error handling.
  def run_sql_plus_plus_query( ctx: Context, bucket_name: str, scope_name: str, query: str ) -> list[dict[str, Any]]: """Run a SQL++ query on a scope and return the results as a list of JSON objects. The query will be run on the specified scope in the specified bucket. The query should use collection names directly without bucket/scope prefixes, as the scope context is automatically set. Example: query = "SELECT * FROM users WHERE age > 18" # Incorrect: "SELECT * FROM bucket.scope.users WHERE age > 18" """ cluster = get_cluster_connection(ctx) bucket = connect_to_bucket(cluster, bucket_name) app_context = ctx.request_context.lifespan_context read_only_query_mode = app_context.read_only_query_mode logger.info(f"Running SQL++ queries in read-only mode: {read_only_query_mode}") try: scope = bucket.scope(scope_name) results = [] # If read-only mode is enabled, check if the query is a data or structure modification query if read_only_query_mode: parsed_query = parse_sqlpp(query) data_modification_query = modifies_data(parsed_query) structure_modification_query = modifies_structure(parsed_query) if data_modification_query: logger.error("Data modification query is not allowed in read-only mode") raise ValueError( "Data modification query is not allowed in read-only mode" ) if structure_modification_query: logger.error( "Structure modification query is not allowed in read-only mode" ) raise ValueError( "Structure modification query is not allowed in read-only mode" ) # Run the query if it is not a data or structure modification query result = scope.query(query) for row in result: results.append(row) return results except Exception as e: logger.error(f"Error running query: {e!s}", exc_info=True) raise
• src/tools/__init__.py:17-21 (registration)

  Import statement that brings get_schema_for_collection into the tools.__init__.py for exposure via ALL_TOOLS.
  # Query tools from .query import ( get_schema_for_collection, run_sql_plus_plus_query, )