MCP Background Job Server

Instructions

Implementation Reference

• src/mcp_background_job/server.py:181-200 (handler)

  FastMCP tool handler for 'kill_job': validates job_id input using Pydantic Field, calls JobManager.kill_job, returns KillOutput with result status.

  @mcp.tool()
  async def kill_job(
      job_id: str = Field(..., description="Job ID to kill"),
  ) -> KillOutput:
      """Kill a running background job.
  
      Args:
          job_id: The UUID of the job to terminate
  
      Returns:
          KillOutput indicating the result of the kill operation
      """
      try:
          job_manager = get_job_manager()
          kill_result = await job_manager.kill_job(job_id)
          return KillOutput(status=kill_result)
      except Exception as e:
          logger.error(f"Error killing job {job_id}: {e}")
          raise ToolError(f"Failed to kill job: {str(e)}")

• src/mcp_background_job/models.py:90-102 (schema)

  Pydantic schemas for kill_job tool: KillInput defines the job_id parameter, KillOutput defines the status response field with possible values.

  class KillInput(BaseModel):
      """Input for kill tool."""
  
      job_id: str = Field(..., description="Job ID to kill")
  
  
  class KillOutput(BaseModel):
      """Output from kill tool."""
  
      status: str = Field(
          ..., description="Kill result: 'killed', 'already_terminated', or 'not_found'"
      )

• src/mcp_background_job/service.py:171-205 (helper)

  Core JobManager.kill_job implementation: checks if job exists and is running, terminates the process using ProcessWrapper.kill(), updates job status and metadata, returns result status string.

  async def kill_job(self, job_id: str) -> str:
      """Kill running job.
  
      Args:
          job_id: Job identifier
  
      Returns:
          Kill result: 'killed', 'already_terminated', or 'not_found'
      """
      if job_id not in self._jobs:
          return "not_found"
  
      job = self._jobs[job_id]
      process_wrapper = self._processes.get(job_id)
  
      # Update status first
      await self._update_job_status(job_id)
  
      if job.status in [JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.KILLED]:
          return "already_terminated"
  
      if process_wrapper is None:
          job.status = JobStatus.FAILED
          return "already_terminated"
  
      # Kill the process
      if process_wrapper.kill():
          job.status = JobStatus.KILLED
          job.completed = datetime.now(timezone.utc)
          job.exit_code = process_wrapper.get_exit_code()
          logger.info(f"Killed job {job_id}")
          return "killed"
      else:
          return "already_terminated"