get_job_status
Check the status of a background job on the MCP server by providing its job ID. Returns running, completed, failed, or killed status for monitoring and management.
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 |
Implementation Reference
- src/mcp_background_job/server.py:50-70 (handler)The main MCP tool handler for 'get_job_status'. It uses @mcp.tool() decorator for registration, defines input schema via Pydantic Field, calls the JobManager helper, and returns StatusOutput.@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 model defining the output schema for the get_job_status tool.class StatusOutput(BaseModel): """Output from status tool.""" status: JobStatus = Field(..., description="Current job status")
- Pydantic model defining the input schema for the get_job_status tool (though inline Field is used in handler).class StatusInput(BaseModel): """Input for status tool.""" job_id: str = Field(..., description="Job ID to check")
- Core helper method in JobManager class that retrieves, updates, and returns the status of a specific job.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
- Enum defining possible job status values used in the tool's output schema.class JobStatus(str, Enum): """Status of a background job.""" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" KILLED = "killed"