list_schemas
Retrieve all schema names in the current CockroachDB database to understand database structure and relationships.
Instructions
List schemas in the current database.
Returns:
List of schemas.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/cockroachdb_mcp/server.py:138-149 (handler)MCP tool handler for 'list_schemas'. Decorated with @mcp.tool(), it calls the helper function from tables module 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)}
- Helper function implementing the logic to list schemas by querying information_schema.schemata, filtering system and disallowed schemas, used by the MCP handler.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)}