MCP Background Job Server

Instructions

Implementation Reference

• src/mcp_background_job/server.py:151-179 (handler)

  MCP tool handler for 'interact_with_job': sends user input to a background job's stdin and returns immediate output, with error handling and proxying to JobManager service.

  @mcp.tool()
  async def interact_with_job(
      job_id: str = Field(..., description="Job ID to interact with"),
      input: str = Field(..., description="Input to send to the job's stdin"),
  ) -> ProcessOutput:
      """Send input to a job's stdin and return any immediate output.
  
      Args:
          job_id: The UUID of the job to interact with
          input: Text to send to the job's stdin
  
      Returns:
          ProcessOutput containing any immediate stdout/stderr output after sending input
      """
      try:
          job_manager = get_job_manager()
          interaction_result = await job_manager.interact_with_job(job_id, input)
          return interaction_result
      except KeyError:
          raise ToolError(f"Job {job_id} not found")
      except RuntimeError as e:
          if "not running" in str(e):
              raise ToolError(f"Job {job_id} is not running and cannot accept input")
          else:
              raise ToolError(f"Failed to interact with job: {str(e)}")
      except Exception as e:
          logger.error(f"Error interacting with job {job_id}: {e}")
          raise ToolError(f"Failed to interact with job: {str(e)}")

• src/mcp_background_job/service.py:253-282 (helper)

  Core JobManager method implementing interact_with_job logic: validates job is running, sends input to process stdin via ProcessWrapper, and returns output.

  async def interact_with_job(self, job_id: str, input_text: str) -> ProcessOutput:
      """Send input to job stdin, return immediate output.
  
      Args:
          job_id: Job identifier
          input_text: Text to send to stdin
  
      Returns:
          ProcessOutput with any immediate stdout/stderr output
  
      Raises:
          KeyError: If job_id doesn't exist
          RuntimeError: If job is not running or stdin not available
      """
      if job_id not in self._jobs:
          raise KeyError(f"Job {job_id} not found")
  
      # Update job status first
      await self._update_job_status(job_id)
  
      job = self._jobs[job_id]
      if job.status != JobStatus.RUNNING:
          raise RuntimeError(f"Job {job_id} is not running (status: {job.status})")
  
      process_wrapper = self._processes.get(job_id)
      if process_wrapper is None:
          raise RuntimeError(f"Process wrapper for job {job_id} not found")
  
      return await process_wrapper.send_input(input_text)