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) )