MCP Background Job Server

Instructions

Implementation Reference

• src/mcp_background_job/server.py:96-120 (handler)

  The primary MCP tool handler for 'tail_job_output', decorated with @mcp.tool() for automatic registration. Handles input validation via Pydantic Fields and error handling, delegating core logic to JobManager.

  @mcp.tool()
  async def tail_job_output(
      job_id: str = Field(..., description="Job ID to tail"),
      lines: int = Field(50, description="Number of lines to return", ge=1, le=1000),
  ) -> ProcessOutput:
      """Get the last N lines of stdout and stderr from a job.
  
      Args:
          job_id: The UUID of the job to tail
          lines: Number of lines to return (1-1000, default 50)
  
      Returns:
          ProcessOutput containing the last N lines of stdout and stderr
      """
      try:
          job_manager = get_job_manager()
          job_output = await job_manager.tail_job_output(job_id, lines)
          return job_output
      except KeyError:
          raise ToolError(f"Job {job_id} not found")
      except ValueError as e:
          raise ToolError(f"Invalid parameter: {str(e)}")
      except Exception as e:
          logger.error(f"Error tailing job output for {job_id}: {e}")
          raise ToolError(f"Failed to tail job output: {str(e)}")

• src/mcp_background_job/models.py:42-47 (schema)

  Pydantic model defining the output schema for tail_job_output, containing stdout and stderr strings.

  class ProcessOutput(BaseModel):
      """Structured stdout/stderr output from a process."""
  
      stdout: str = Field(..., description="Standard output content")
      stderr: str = Field(..., description="Standard error content")

• src/mcp_background_job/service.py:227-252 (helper)

  JobManager.tail_job_output helper method that validates job existence and lines parameter, then delegates to ProcessWrapper.tail_output.

  async def tail_job_output(self, job_id: str, lines: int) -> ProcessOutput:
      """Get last N lines of output.
  
      Args:
          job_id: Job identifier
          lines: Number of lines to return
  
      Returns:
          ProcessOutput with last N lines of stdout and stderr
  
      Raises:
          KeyError: If job_id doesn't exist
          ValueError: If lines is not positive
      """
      if job_id not in self._jobs:
          raise KeyError(f"Job {job_id} not found")
  
      if lines <= 0:
          raise ValueError("Number of lines must be positive")
  
      process_wrapper = self._processes.get(job_id)
      if process_wrapper is None:
          return ProcessOutput(stdout="", stderr="")
  
      return process_wrapper.tail_output(lines)

• src/mcp_background_job/process.py:264-280 (helper)

  ProcessWrapper.tail_output core implementation that thread-safely extracts the last N lines from stdout and stderr ring buffers (deques).

  def tail_output(self, lines: int) -> ProcessOutput:
      """Get last N lines of output.
  
      Args:
          lines: Number of lines to return from the end
  
      Returns:
          ProcessOutput containing last N lines of stdout and stderr
      """
      with self._buffer_lock:
          # Get last N lines from each buffer
          stdout_lines = list(self.stdout_buffer)[-lines:] if lines > 0 else []
          stderr_lines = list(self.stderr_buffer)[-lines:] if lines > 0 else []
  
      return ProcessOutput(
          stdout="\n".join(stdout_lines), stderr="\n".join(stderr_lines)
      )

• src/mcp_background_job/models.py:71-76 (schema)

  Pydantic model defining the input schema for tail_job_output tool (though inline Fields are used in handler).

  class TailInput(BaseModel):
      """Input for tail tool."""
  
      job_id: str = Field(..., description="Job ID to tail")
      lines: int = Field(50, description="Number of lines to return", ge=1, le=1000)