CockroachDB MCP Server

Instructions

Args:
    job_type: Filter by job type (BACKUP, RESTORE, IMPORT, etc.).
    status: Filter by status (running, succeeded, failed, etc.).
    limit: Maximum jobs to return.

Returns:
    List of jobs.

Implementation Reference

• src/cockroachdb_mcp/tools/cluster.py:224-286 (handler)

  Core handler function that executes SQL query on crdb_internal.jobs with optional filters for job_type, status, and limit. Processes results into formatted job objects and returns them or an error.

  async def show_jobs(
      job_type: str | None = None,
      status: str | None = None,
      limit: int = 20,
  ) -> dict[str, Any]:
      """Show background jobs in the cluster.
  
      Args:
          job_type: Filter by job type (BACKUP, RESTORE, IMPORT, etc.).
          status: Filter by status (running, succeeded, failed, etc.).
          limit: Maximum jobs to return.
  
      Returns:
          List of jobs.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          query = """
              SELECT
                  job_id,
                  job_type,
                  status,
                  description,
                  created,
                  started,
                  finished,
                  fraction_completed,
                  error
              FROM crdb_internal.jobs
              WHERE 1=1
          """
  
          if job_type:
              query += f" AND job_type = '{job_type.upper()}'"
          if status:
              query += f" AND status = '{status.lower()}'"
  
          query += f" ORDER BY created DESC LIMIT {limit}"
  
          async with conn.cursor() as cur:
              await cur.execute(query)
              rows = await cur.fetchall()
  
          jobs = []
          for row in rows:
              jobs.append(
                  {
                      "job_id": str(row.get("job_id")),
                      "job_type": row.get("job_type"),
                      "status": row.get("status"),
                      "description": row.get("description"),
                      "created": str(row.get("created")) if row.get("created") else None,
                      "started": str(row.get("started")) if row.get("started") else None,
                      "finished": str(row.get("finished")) if row.get("finished") else None,
                      "progress": row.get("fraction_completed"),
                      "error": row.get("error"),
                  }
              )
  
          return {"jobs": jobs, "count": len(jobs)}
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:464-483 (registration)

  MCP tool registration via @mcp.tool() decorator in the FastMCP server. This wrapper delegates execution to the cluster.show_jobs handler and adds error handling.

  @mcp.tool()
  async def show_jobs(
      job_type: str | None = None,
      status: str | None = None,
      limit: int = 20,
  ) -> dict[str, Any]:
      """Show background jobs in the cluster.
  
      Args:
          job_type: Filter by job type (BACKUP, RESTORE, IMPORT, etc.).
          status: Filter by status (running, succeeded, failed, etc.).
          limit: Maximum jobs to return.
  
      Returns:
          List of jobs.
      """
      try:
          return await cluster.show_jobs(job_type, status, limit)
      except Exception as e:
          return {"status": "error", "error": str(e)}