CockroachDB MCP Server

Instructions

Args:
    limit: Maximum statements to return.

Returns:
    List of active statements.

Implementation Reference

• src/cockroachdb_mcp/server.py:502-515 (handler)

  MCP tool handler for 'show_statements'. Registered via @mcp.tool() decorator. Thin wrapper that calls the underlying cluster.show_statements() and handles 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 (helper)

  Core helper function implementing the logic to show active statements. Connects to DB, queries crdb_internal.cluster_queries, processes results into structured dict.

  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:502-502 (registration)

  The @mcp.tool() decorator registers the show_statements function as an MCP tool.

  @mcp.tool()