CouchDB MCP Server

Instructions

Implementation Reference

• couchdb_mcp_server.py:416-447 (handler)

  Main handler implementation for couchdb_search_documents that executes Mango queries using db.find() method with selector, limit, and skip parameters

  async def _search_documents(self, database: str, query: dict, limit: int, skip: int) -> list[TextContent]:
      """Search documents using Mango query."""
      try:
          db = self._get_server()[database]
  
          # Build Mango query using the find() method
          mango_query = {
              "selector": query,
              "limit": limit,
              "skip": skip
          }
  
          # Use the db.find() method (available in CouchDB >= 2.0)
          docs = list(db.find(mango_query))
  
          response = {
              "docs": docs,
              "count": len(docs)
          }
  
          # If no results, provide helpful suggestion
          if len(docs) == 0:
              response["note"] = "No documents matched the query. To verify documents exist, use couchdb_list_documents with include_docs=true"
  
          return [TextContent(type="text", text=json.dumps(response, indent=2))]
      except KeyError:
          return [TextContent(type="text", text=f"Database '{database}' not found")]
      except AttributeError:
          # Fallback to REST API if find() method not available
          return await self._search_documents_fallback(database, query, limit, skip)
      except Exception as e:
          return [TextContent(type="text", text=f"Error searching documents: {str(e)}")]

• couchdb_mcp_server.py:449-479 (handler)

  Fallback handler that uses raw REST API (_find endpoint) when db.find() method is not available

  async def _search_documents_fallback(self, database: str, query: dict, limit: int, skip: int) -> list[TextContent]:
      """Fallback search using raw REST API."""
      try:
          db = self._get_server()[database]
  
          mango_query = {
              "selector": query,
              "limit": limit,
              "skip": skip
          }
  
          # Make a request to the _find endpoint
          result = db.resource.post_json('_find', body=mango_query)
  
          docs = result[1].get("docs", [])
          warning = result[1].get("warning", None)
  
          response = {
              "docs": docs,
              "count": len(docs)
          }
  
          if warning:
              response["warning"] = warning
  
          if len(docs) == 0:
              response["note"] = "No documents matched the query. To verify documents exist, use couchdb_list_documents with include_docs=true"
  
          return [TextContent(type="text", text=json.dumps(response, indent=2))]
      except Exception as e:
          return [TextContent(type="text", text=f"Error in fallback search: {str(e)}")]

• couchdb_mcp_server.py:175-200 (schema)

  Tool schema definition that defines input parameters (database, query, limit, skip) and their types for couchdb_search_documents

  Tool(
      name="couchdb_search_documents",
      description="Search for documents in a database using a Mango query. Works without indexes but creating indexes (via couchdb_create_index) improves performance significantly.",
      inputSchema={
          "type": "object",
          "properties": {
              "database": {
                  "type": "string",
                  "description": "Name of the database",
              },
              "query": {
                  "type": "object",
                  "description": "Mango query selector (e.g., {'name': 'John'} for exact match, {'age': {'$gt': 18}} for comparisons)",
              },
              "limit": {
                  "type": "integer",
                  "description": "Maximum number of documents to return (default: 25)",
              },
              "skip": {
                  "type": "integer",
                  "description": "Number of documents to skip (default: 0)",
              },
          },
          "required": ["database", "query"],
      },
  ),

• couchdb_mcp_server.py:295-301 (registration)

  Registration of the tool handler that routes couchdb_search_documents calls to the _search_documents method with argument extraction

  elif name == "couchdb_search_documents":
      return await self._search_documents(
          arguments["database"],
          arguments["query"],
          arguments.get("limit", 25),
          arguments.get("skip", 0)
      )