CockroachDB MCP Server

Instructions

Returns:
    List of schemas.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:71-115 (handler)

  Core handler function that executes the logic to list schemas by querying information_schema.schemata, filtering system schemas and using allowed schema configuration.

  async def list_schemas(database: str | None = None) -> dict[str, Any]:
      """List schemas in a database.
  
      Args:
          database: Database name (uses current if not specified).
  
      Returns:
          List of schemas.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          async with conn.cursor() as cur:
              await cur.execute("""
                  SELECT schema_name
                  FROM information_schema.schemata
                  WHERE catalog_name = current_database()
                  ORDER BY schema_name
              """)
              rows = await cur.fetchall()
  
          schemas = []
          system_schemas = {"crdb_internal", "information_schema", "pg_catalog", "pg_extension"}
  
          for row in rows:
              schema_name = row.get("schema_name", "")
  
              # Check if allowed
              if not _is_allowed_schema(schema_name):
                  continue
  
              schemas.append(
                  {
                      "name": schema_name,
                      "is_system": schema_name in system_schemas,
                  }
              )
  
          return {
              "schemas": schemas,
              "count": len(schemas),
              "database": database or connection_manager.current_database,
          }
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:138-148 (registration)

  MCP tool registration using @mcp.tool() decorator. Thin wrapper that calls the core handler in tables.list_schemas() and handles exceptions.

  @mcp.tool()
  async def list_schemas() -> dict[str, Any]:
      """List schemas in the current database.
  
      Returns:
          List of schemas.
      """
      try:
          return await tables.list_schemas()
      except Exception as e:
          return {"status": "error", "error": str(e)}