CockroachDB MCP Server

Instructions

Implementation Reference

• src/cockroachdb_mcp/server.py:502-516 (registration)

  Registers the 'show_statements' MCP tool with @mcp.tool() decorator. Thin handler wrapper that delegates to cluster.show_statements, catching exceptions.
  @mcp.tool() async def show_statements(limit: int = 20) -> dict[str, Any]: """Show active statements in the cluster. Args: limit: Maximum statements to return. Returns: List of active statements. """ try: return await cluster.show_statements(limit) except Exception as e: return {"status": "error", "error": str(e)}
• src/cockroachdb_mcp/tools/cluster.py:341-385 (handler)

  Core implementation of show_statements: executes SQL query on crdb_internal.cluster_queries, processes rows into structured statements list, handles errors.
  async def show_statements(limit: int = 20) -> dict[str, Any]: """Show active statements in the cluster. Args: limit: Maximum statements to return. Returns: List of active statements. """ conn = await connection_manager.ensure_connected() try: async with conn.cursor() as cur: await cur.execute(f""" SELECT query_id, node_id, user_name, query, start, phase, application_name FROM crdb_internal.cluster_queries ORDER BY start DESC LIMIT {limit} """) rows = await cur.fetchall() statements = [] for row in rows: statements.append( { "query_id": row.get("query_id"), "node_id": row.get("node_id"), "user": row.get("user_name"), "query": row.get("query"), "started": str(row.get("start")) if row.get("start") else None, "phase": row.get("phase"), "application": row.get("application_name"), } ) return {"statements": statements, "count": len(statements)} except Exception as e: return {"status": "error", "error": str(e)}
• src/cockroachdb_mcp/server.py:503-503 (schema)

  Input schema defined by function signature: limit: int = 20, output: dict[str, Any] (in MCP tool registration). Similar in helper.
  async def show_statements(limit: int = 20) -> dict[str, Any]: