get_job_status
Check the current status of a background job by providing its UUID to monitor progress and determine if it's running, completed, failed, or killed.
Instructions
Get the current status of a background job.
Args: job_id: The UUID of the job to check
Returns: The current status of the job (running, completed, failed, or killed)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| job_id | Yes | Job ID to check |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| status | Yes | Current job status |
Implementation Reference
- src/mcp_background_job/server.py:50-71 (handler)FastMCP tool handler implementation for 'get_job_status'. Calls JobManager.get_job_status(job_id) and returns StatusOutput(status=job_status), handling errors with ToolError.
@mcp.tool() async def get_job_status( job_id: str = Field(..., description="Job ID to check"), ) -> StatusOutput: """Get the current status of a background job. Args: job_id: The UUID of the job to check Returns: The current status of the job (running, completed, failed, or killed) """ try: job_manager = get_job_manager() job_status = await job_manager.get_job_status(job_id) return StatusOutput(status=job_status) except KeyError: raise ToolError(f"Job {job_id} not found") except Exception as e: logger.error(f"Error getting job status for {job_id}: {e}") raise ToolError(f"Failed to get job status: {str(e)}") - Pydantic BaseModel StatusOutput defining the output schema: status: JobStatus.
class StatusOutput(BaseModel): """Output from status tool.""" status: JobStatus = Field(..., description="Current job status") - JobStatus enum defining possible job status values used in StatusOutput.
class JobStatus(str, Enum): """Status of a background job.""" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" KILLED = "killed" - JobManager.get_job_status method providing core logic: checks job existence, updates status via _update_job_status, returns current status.
async def get_job_status(self, job_id: str) -> JobStatus: """Get current status of job. Args: job_id: Job identifier Returns: Current job status Raises: KeyError: If job_id doesn't exist """ if job_id not in self._jobs: raise KeyError(f"Job {job_id} not found") # Update job status from process await self._update_job_status(job_id) return self._jobs[job_id].status