MCP Background Job Server

Instructions

Implementation Reference

• src/mcp_background_job/server.py:73-94 (handler)

  The main handler function for the 'get_job_output' MCP tool, decorated with @mcp.tool() for registration. Delegates to JobManager service.
  @mcp.tool() async def get_job_output( job_id: str = Field(..., description="Job ID to get output from"), ) -> ProcessOutput: """Get the complete stdout and stderr output of a job. Args: job_id: The UUID of the job to get output from Returns: ProcessOutput containing the complete stdout and stderr content """ try: job_manager = get_job_manager() job_output = await job_manager.get_job_output(job_id) return job_output except KeyError: raise ToolError(f"Job {job_id} not found") except Exception as e: logger.error(f"Error getting job output for {job_id}: {e}") raise ToolError(f"Failed to get job output: {str(e)}")
• src/mcp_background_job/service.py:206-226 (helper)

  Core implementation in JobManager class that retrieves the process output from the stored process wrapper.
  async def get_job_output(self, job_id: str) -> ProcessOutput: """Get full stdout/stderr output. Args: job_id: Job identifier Returns: ProcessOutput with complete stdout and stderr Raises: KeyError: If job_id doesn't exist """ if job_id not in self._jobs: raise KeyError(f"Job {job_id} not found") process_wrapper = self._processes.get(job_id) if process_wrapper is None: return ProcessOutput(stdout="", stderr="") return process_wrapper.get_output()
• src/mcp_background_job/models.py:42-47 (schema)

  Pydantic model used for input validation and output typing of the get_job_output tool.
  class ProcessOutput(BaseModel): """Structured stdout/stderr output from a process.""" stdout: str = Field(..., description="Standard output content") stderr: str = Field(..., description="Standard error content")