### Install MCP Background Job Server Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Installs and runs the MCP Background Job Server using the `uvx` package manager. This is the recommended quick installation method. ```bash # Install and run the MCP server uvx mcp-background-job ``` -------------------------------- ### Development Server Workflow Example (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This bash script illustrates a typical development workflow for managing a background job, such as starting a development server. It shows how to execute a command, monitor its output, check its status, and eventually terminate the job. ```bash # Start a development server job_id=$(echo '{"command": "npm run dev"}' | mcp-tool execute) # Monitor the startup mcp-tool tail --job_id "$job_id" --lines 10 # Check if server is ready mcp-tool status --job_id "$job_id" # Stop the server mcp-tool kill --job_id "$job_id" ``` -------------------------------- ### Interactive Process Example (Python REPL) (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This bash script shows how to run an interactive process, like a Python REPL, as a background job. It demonstrates starting the REPL, sending commands to its standard input, and receiving the output. ```bash # Start Python REPL job_id=$(echo '{"command": "python -i"}' | mcp-tool execute) # Send Python code mcp-tool interact --job_id "$job_id" --input "print('Hello, World!')\n" # Send more commands mcp-tool interact --job_id "$job_id" --input "import sys; print(sys.version)\n" # Exit REPL mcp-tool interact --job_id "$job_id" --input "exit()\n" ``` -------------------------------- ### Set Up MCP Background Job Server for Development Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Provides steps for setting up the project locally, including cloning the repository, installing dependencies using `uv`, and installing the package in development mode. ```bash git clone https://github.com/dylan-gluck/mcp-background-job.git cd mcp-background-job uv sync uv add -e . ``` -------------------------------- ### Run Tests After Installation (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Installs the package and its dependencies, then executes the test suite using pytest. This is necessary when tests are not running, ensuring the environment is correctly set up. ```bash uv sync uv add -e . uv run pytest tests/ ``` -------------------------------- ### Long-running Build Process Example (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This bash script demonstrates handling a long-running build process as a background job. It includes starting the build, monitoring its progress periodically, and retrieving the final output upon completion or failure. ```bash # Start a build process job_id=$(echo '{"command": "docker build -t myapp ."}' | mcp-tool execute) # Monitor build progress while true; do status=$(mcp-tool status --job_id "$job_id") if [[ "$status" != "running" ]]; then break; fi mcp-tool tail --job_id "$job_id" --lines 5 sleep 10 done # Get final build output mcp-tool output --job_id "$job_id" ``` -------------------------------- ### Install Package in Development Mode (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Ensures the package is installed in development mode, which is crucial for resolving import errors. This command should be run within the project's environment. ```bash uv add -e . ``` -------------------------------- ### Execute Command with MCP Background Job Server (Python) Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Starts a new background job by executing a given shell command. It returns a unique job ID for tracking the process lifecycle. Supports various commands like starting development servers, build processes, or test suites. Handles potential errors such as empty commands. ```python from mcp_background_job.server import execute_command # Start a development server result = await execute_command(command="npm run dev") print(f"Job started with ID: {result.job_id}") # Output: Job started with ID: 550e8400-e29b-41d4-a716-446655440000 # Start a build process build_job = await execute_command(command="docker build -t myapp .") print(f"Build job ID: {build_job.job_id}") # Start a long-running test suite test_job = await execute_command(command="pytest tests/ -v --slow") print(f"Tests running with job ID: {test_job.job_id}") # Handle errors try: job = await execute_command(command="") except ToolError as e: print(f"Error: {e}") # Invalid command: Command cannot be empty # Returns: ExecuteOutput(job_id="uuid-string") ``` -------------------------------- ### ProcessWrapper: Low-level Process Management in Python Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Manages background processes with I/O handling and buffering. This includes creating, starting, monitoring status, sending input, retrieving output (full and tailed), killing processes, and cleaning up resources. It requires `ProcessWrapper` and `JobStatus` from `mcp_background_job.process` and `mcp_background_job.models` respectively. ```python from mcp_background_job.process import ProcessWrapper from mcp_background_job.models import JobStatus job_id = "test-job-123" wrapper = ProcessWrapper( job_id=job_id, command="python long_running_script.py", max_output_size=10 * 1024 * 1024 # 10MB ) await wrapper.start() print(f"Process started with PID: {wrapper.get_pid()}") print(f"Started at: {wrapper.started_at}") status = wrapper.get_status() print(f"Status: {status}") result = await wrapper.send_input("input_data\n") print(f"Response: {result.stdout}") output = wrapper.get_output() print(f"Full stdout: {output.stdout}") print(f"Full stderr: {output.stderr}") tail = wrapper.tail_output(50) print(f"Recent output: {tail.stdout}") if wrapper.kill(): print("Process terminated") print(f"Exit code: {wrapper.get_exit_code()}") print(f"Completed at: {wrapper.completed_at}") wrapper.cleanup() ``` -------------------------------- ### Async Command Execution in Python Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md This Python code demonstrates an excellent example of asynchronous command execution using `async/await`. It emphasizes clean implementation and proper error handling, contributing to the project's robustness. ```python async def execute_command(self, command: str) -> str: # Clean async implementation with proper error handling ``` -------------------------------- ### Run MCP Background Job Server Locally Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Commands to run the MCP Background Job Server directly from the command line, either using `uvx` for standard installation or `uv run` for development purposes. ```bash # Run with stdio transport (most common) uvx mcp-background-job # Or for development: uv run python -m mcp_background_job ``` -------------------------------- ### Interactive Tools API Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md API endpoints for interacting with background jobs, including starting, sending input, and terminating jobs. ```APIDOC ## POST /execute ### Description Starts a new background job. ### Method POST ### Endpoint /execute ### Parameters #### Request Body - **command** (str) - Required - The command to execute in the background job. ### Request Example ```json { "command": "npm run dev" } ``` ### Response #### Success Response (200) - **job_id** (str) - The unique identifier for the started job. #### Response Example ```json { "job_id": "some-unique-job-id" } ``` ## POST /interact ### Description Sends input to the standard input of a running job. ### Method POST ### Endpoint /interact ### Parameters #### Request Body - **job_id** (str) - Required - The ID of the job to interact with. - **input** (str) - Required - The input string to send to the job's stdin. ### Request Example ```json { "job_id": "some-unique-job-id", "input": "print('Hello, World!')\n" } ``` ### Response #### Success Response (200) - **stdout** (str) - The standard output from the job after processing the input. - **stderr** (str) - The standard error output from the job after processing the input. #### Response Example ```json { "stdout": "Hello, World!\n", "stderr": "" } ``` ## POST /kill ### Description Terminates a running background job. ### Method POST ### Endpoint /kill ### Parameters #### Request Body - **job_id** (str) - Required - The ID of the job to terminate. ### Request Example ```json { "job_id": "some-unique-job-id" } ``` ### Response #### Success Response (200) - **status** (str) - The final status of the job after termination (e.g., 'killed'). #### Response Example ```json { "status": "killed" } ``` ``` -------------------------------- ### List All Jobs Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieve a list of all background jobs with their current status, sorted by start time. Allows filtering by job status. ```APIDOC ## List All Jobs ### Description Retrieve a list of all background jobs with their current status. Jobs are sorted by start time (newest first). ### Method GET ### Endpoint `/jobs` ### Query Parameters - **status** (string) - Optional - Filter jobs by status (e.g., 'running', 'completed', 'failed'). ### Response #### Success Response (200) - **jobs** (array[object]) - A list of job summaries. - **job_id** (string) - Unique identifier for the job. - **status** (string) - Current status of the job (e.g., 'running', 'completed', 'failed', 'killed'). - **command** (string) - The command executed by the job. - **started** (string) - Timestamp when the job started. #### Response Example ```json { "jobs": [ { "job_id": "550e8400-e29b-41d4-a716-446655440000", "status": "running", "command": "npm run dev", "started": "2025-12-02 10:30:45.123456+00:00" }, { "job_id": "661f9510-f3ac-52e5-b827-557766551111", "status": "completed", "command": "pytest tests/", "started": "2025-12-02 10:25:12.789012+00:00" } ] } ``` ``` -------------------------------- ### Execute, Monitor, and Manage Background Jobs with Python Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md A Python example showcasing basic interactions with the MCP Background Job Server. It demonstrates executing a command, checking its status, retrieving output, sending input, and terminating the job. ```python # 1. Execute a long-running command execute_result = await execute_command("npm run dev") job_id = execute_result.job_id # 2. Check job status status = await get_job_status(job_id) print(f"Job status: {status.status}") # 3. Get recent output output = await tail_job_output(job_id, lines=20) print("Recent output:", output.stdout) # 4. Interact with the process interaction = await interact_with_job(job_id, "some input\n") print("Process response:", interaction.stdout) # 5. Kill the job when done result = await kill_job(job_id) print(f"Kill result: {result.status}") ``` -------------------------------- ### Pydantic Model for Background Job Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md An example of a type-safe data structure for a background job using Pydantic. It includes fields for job ID, command, status, and timestamps, with descriptions and type validation enforced by Pydantic. ```python from pydantic import BaseModel, Field from datetime import datetime from enum import Enum class JobStatus(str, Enum): PENDING = "pending" RUNNING = "running" COMPLETED = "completed" FAILED = "failed" class BackgroundJob(BaseModel): job_id: str = Field(..., description="UUID v4 job identifier") command: str = Field(..., description="Shell command being executed") status: JobStatus = Field(..., description="Current job status") started: datetime = Field(..., description="UTC timestamp when job started") ``` -------------------------------- ### Execute Command Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Starts a new background job by executing a shell command. It returns a unique job ID that can be used to track the job's progress and retrieve its output. ```APIDOC ## POST /execute_command ### Description Executes a shell command in a new background job. ### Method POST ### Endpoint /execute_command ### Parameters #### Request Body - **command** (string) - Required - The shell command to execute. ### Request Example ```json { "command": "npm run dev" } ``` ### Response #### Success Response (200) - **job_id** (string) - The unique identifier for the background job. #### Response Example ```json { "job_id": "550e8400-e29b-41d4-a716-446655440000" } ``` #### Error Handling - **ToolError**: Raised if the command is empty or invalid. ``` -------------------------------- ### Error Handling Example (Python) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Demonstrates structured error handling for background job operations using custom exceptions derived from ToolError. This ensures consistent error reporting across different job management functions. ```python from fastmcp.exceptions import ToolError # Example usage within a tool: # if job_id not in self._jobs: # raise ToolError(f"Job {job_id} not found") ``` -------------------------------- ### Get Complete Job Output Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieves the complete standard output (stdout) and standard error (stderr) from a completed background job. This is useful for analyzing the results of a finished process. ```APIDOC ## GET /jobs/{job_id}/output ### Description Retrieves the full stdout and stderr output of a background job. ### Method GET ### Endpoint /jobs/{job_id}/output ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier of the job. ### Response #### Success Response (200) - **stdout** (string) - The standard output of the job. - **stderr** (string) - The standard error of the job. #### Response Example ```json { "stdout": "Build successful!", "stderr": "" } ``` #### Error Handling - **ToolError**: Raised if the job ID is not found. ``` -------------------------------- ### Get Job Details and Statistics in Python Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieves detailed information about a specific job, including its status, PID, and exit code. It also fetches overall job statistics like total, running, completed, failed, and killed jobs. Finally, it demonstrates how to clean up completed jobs and perform a graceful shutdown of the manager. ```python job = await manager.get_job(job_id) print(f"Job: {job.command}") print(f"Status: {job.status}") print(f"PID: {job.pid}") print(f"Started: {job.started}") print(f"Exit code: {job.exit_code}") stats = manager.get_stats() print(f"Total jobs: {stats['total']}") print(f"Running: {stats['running']}") print(f"Completed: {stats['completed']}") print(f"Failed: {stats['failed']}") print(f"Killed: {stats['killed']}") cleanup_count = manager.cleanup_completed_jobs() print(f"Cleaned up {cleanup_count} jobs") await manager.shutdown() ``` -------------------------------- ### Python ProcessWrapper for Handling Child Processes Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Provides an abstraction layer for managing individual child processes. The `ProcessWrapper` class handles starting processes, capturing their I/O streams (stdout, stderr, stdin), and providing methods to check status, kill the process, and retrieve output. ```python import subprocess from typing import Optional, List class ProcessWrapper: def __init__(self, job_id: str, command: str): self.job_id = job_id self.command = command self.process: Optional[subprocess.Popen] = None self.stdout_buffer: List[str] = [] self.stderr_buffer: List[str] = [] async def start(self): """Start the process with proper I/O handling""" async def send_input(self, text: str) -> ProcessOutput: """Send input to process stdin""" def get_status(self) -> JobStatus: """Get current process status""" def kill(self) -> bool: """Kill the process""" def get_output(self) -> ProcessOutput: """Get all captured output""" def tail_output(self, lines: int) -> ProcessOutput: """Get last N lines of output""" ``` -------------------------------- ### Get Complete Job Output with MCP Background Job Server (Python) Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Fetches the entire standard output (stdout) and standard error (stderr) streams from a completed background job. This is useful for analyzing the full results or debugging issues. Output is buffered and may be truncated if buffer limits are reached. ```python from mcp_background_job.server import get_job_output job_id = "550e8400-e29b-41d4-a716-446655440000" # Get all output from a completed job output = await get_job_output(job_id=job_id) print("Standard Output:") print(output.stdout) print("\nStandard Error:") print(output.stderr) # Example with build process build_job = await execute_command(command="npm run build") await asyncio.sleep(5) # Wait for build to complete build_output = await get_job_output(job_id=build_job.job_id) if "error" in build_output.stderr.lower(): print("Build failed with errors:") print(build_output.stderr) else: print("Build successful!") print(build_output.stdout) # Output is buffered with max 10MB per stream by default # Older output may be truncated if buffer limit is reached # Returns: ProcessOutput(stdout=str, stderr=str) ``` -------------------------------- ### Configure MCP Background Job via JSON Settings Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Defines configuration for the mcp-background-job server within a JSON structure, specifying the command to run ('uvx'), arguments, and environment variables. This allows for declarative configuration management, similar to environment variables but within a structured JSON file. ```json { "mcpServers": { "background-job": { "command": "uvx", "args": ["mcp-background-job"], "env": { "MCP_BG_MAX_JOBS": "20", "MCP_BG_MAX_OUTPUT_SIZE": "20MB", "MCP_BG_JOB_TIMEOUT": "7200", "MCP_BG_CLEANUP_INTERVAL": "600", "MCP_BG_ALLOWED_COMMANDS": "^npm ,^python ,^pytest " } } } } ``` -------------------------------- ### Get Job Output Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Retrieves the complete standard output and standard error for a specific background job. ```APIDOC ## GET /jobs/{job_id}/output ### Description Gets the complete stdout and stderr output of a background job. ### Method GET ### Endpoint /jobs/{job_id}/output ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier for the job. ### Request Example ``` GET /jobs/a1b2c3d4-e5f6-7890-1234-567890abcdef/output ``` ### Response #### Success Response (200) - **stdout** (string) - The standard output of the job. - **stderr** (string) - The standard error output of the job. #### Response Example ```json { "stdout": "Job output line 1\nJob output line 2", "stderr": "Error message if any" } ``` ``` -------------------------------- ### Get Job Status Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Retrieves the current status of a specific background job identified by its job ID. ```APIDOC ## GET /jobs/{job_id}/status ### Description Gets the current status of a specific background job. ### Method GET ### Endpoint /jobs/{job_id}/status ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier for the job. ### Request Example ``` GET /jobs/a1b2c3d4-e5f6-7890-1234-567890abcdef/status ``` ### Response #### Success Response (200) - **status** (string) - The current status of the job (e.g., 'running', 'completed', 'failed'). - **job_id** (string) - The unique identifier for the job. #### Response Example ```json { "job_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef", "status": "running" } ``` ``` -------------------------------- ### Get Job Status (Python) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Fetches the current status of a specific background job identified by its job ID. This is a read-only, idempotent operation. ```python from fastmcp import mcp from pydantic import Field class StatusOutput: pass # Placeholder for actual output structure @mcp.tool() async def status(job_id: str = Field(..., description="Job ID to check")) -> StatusOutput: """Get the current status of a background job.""" pass ``` -------------------------------- ### Python: Secure Command Execution with subprocess Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md Illustrates a Python code snippet using subprocess.Popen for command execution, highlighting potential shell injection vulnerabilities even with shlex.split. It suggests implementing allowlist/blocklist patterns for enhanced security. Dependencies include the subprocess module. ```python import subprocess import shlex # Inside a class or function # self.command is a string like "ls -l | grep important" args = shlex.split(self.command) self.process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # Potential vulnerability example: # malicious_command = "rm -rf /" # args = shlex.split(malicious_command) # subprocess.Popen(args, ...) ``` -------------------------------- ### Get Job Output (Python) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Retrieves the complete standard output and standard error streams for a specified job ID. This is a read-only, idempotent operation. ```python from fastmcp import mcp from pydantic import Field class ProcessOutput: pass # Placeholder for actual output structure @mcp.tool() async def output(job_id: str = Field(..., description="Job ID to get output from")) -> ProcessOutput: """Get the complete stdout and stderr output of a job.""" pass ``` -------------------------------- ### Project Structure Overview Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md This outlines the modular architecture of the mcp-background-job project, highlighting key components and their responsibilities. It follows a Repository + Service + Controller pattern with Pydantic models for data structures. ```plaintext src/mcp_background_job/ ├── models.py # 121 lines - Clean Pydantic models ├── config.py # 120 lines - Comprehensive configuration ├── process.py # 331 lines - Process management layer ├── service.py # 401 lines - Core business logic └── server.py # 252 lines - FastMCP integration ``` -------------------------------- ### Get Job Status Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieves the current status of a background job using its unique job ID. The status indicates whether the job is running, completed, failed, or killed. ```APIDOC ## GET /jobs/{job_id}/status ### Description Retrieves the status of a specific background job. ### Method GET ### Endpoint /jobs/{job_id}/status ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier of the job. ### Response #### Success Response (200) - **status** (string) - The current status of the job (e.g., RUNNING, COMPLETED, FAILED, KILLED). #### Response Example ```json { "status": "running" } ``` #### Error Handling - **ToolError**: Raised if the job ID is not found. ``` -------------------------------- ### Initialize JobManager Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Instantiate the `JobManager` which serves as the central service for managing the lifecycle of background jobs. It allows for executing commands with custom configurations related to job limits, timeouts, and allowed command patterns. ```python from mcp_background_job.service import JobManager from mcp_background_job.config import BackgroundJobConfig # Create job manager with custom configuration config = BackgroundJobConfig( max_concurrent_jobs=20, max_output_size_bytes=20 * 1024 * 1024, # 20MB default_job_timeout=3600, # 1 hour cleanup_interval_seconds=600, # 10 minutes allowed_command_patterns=["^npm ", "^python ", "^pytest "] ) manager = JobManager(config) # Execute commands job_id = await manager.execute_command("npm test") ``` -------------------------------- ### Python: Validate Command Input Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md Demonstrates a Python code snippet for validating command input, checking for emptiness and whitespace. It points out the missing integration with security patterns defined in configuration, which is crucial for preventing unauthorized command execution. This snippet is part of the service logic. ```python # In src/mcp_background_job/service.py def execute_command(self, command: str): if not command or not command.strip(): raise ValueError("Command cannot be empty") # Missing validation against allowed_command_patterns from config # Example: _validate_command_security(command) # ... proceed with execution ... ``` -------------------------------- ### Python: Exception Handling Practices Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md Illustrates inconsistent exception handling in Python, showing both a generic `except: pass` and a specific `except Exception as e:` with logging. This highlights the need for standardized, specific exception handling and informative logging for better debugging and error management. ```python # Inconsistent handling example 1: # except: # pass # Inconsistent handling example 2: # except Exception as e: # logger.warning(f"Failed to update status: {e}") ``` -------------------------------- ### Configure MCP Background Job via Environment Variables (Bash) Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Sets environment variables to configure the behavior of the mcp-background-job server, such as the maximum number of jobs, output size, job timeout, cleanup interval, working directory, and allowed command patterns. After setting the variables, the server is launched using `uvx mcp-background-job`. ```bash export MCP_BG_MAX_JOBS=20 export MCP_BG_MAX_OUTPUT_SIZE=20MB export MCP_BG_JOB_TIMEOUT=7200 export MCP_BG_CLEANUP_INTERVAL=600 export MCP_BG_WORKING_DIR=/path/to/project export MCP_BG_ALLOWED_COMMANDS="^npm ,^python ,^pytest " uvx mcp-background-job ``` -------------------------------- ### Configure MCP Background Job Server for Claude Code Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Demonstrates how to add the MCP Background Job Server to Claude Code's configuration, either through the desktop application settings or by modifying the configuration file. Requires restarting Claude Code. ```json { "mcpServers": { "background-job": { "command": "uvx", "args": ["mcp-background-job"] } } } ``` -------------------------------- ### Programmatic Configuration of Background Job in Python Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Creates and validates configuration settings for the background job system using Python objects. It allows setting parameters like `max_concurrent_jobs`, `max_output_size_bytes`, `default_job_timeout`, `cleanup_interval_seconds`, `allowed_command_patterns`, and `working_directory`. Configuration can also be loaded from environment variables or a combination of environment and default values. ```python from mcp_background_job.config import BackgroundJobConfig, load_config config = BackgroundJobConfig( max_concurrent_jobs=15, max_output_size_bytes=50 * 1024 * 1024, # 50MB default_job_timeout=3600, cleanup_interval_seconds=300, allowed_command_patterns=["^npm ", "^yarn ", "^python "], working_directory="/path/to/project" ) print(f"Max jobs: {config.max_concurrent_jobs}") print(f"Max output: {config.max_output_size_bytes} bytes") print(f"Timeout: {config.default_job_timeout}s") env_config = BackgroundJobConfig.from_environment() config = load_config() ``` -------------------------------- ### MCP Tool Definition for Command Execution Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md This Python snippet demonstrates how to define an MCP tool for executing commands. It uses the `@mcp.tool()` decorator and Pydantic's `Field` for parameter validation and description, returning an `ExecuteOutput` object. ```python from mcp import mcp from pydantic import Field # Assuming ExecuteOutput is defined elsewhere # class ExecuteOutput(...): # ... @mcp.tool() async def execute_command( command: str = Field(..., description="Shell command to execute"), ) -> ExecuteOutput: """Execute a command as background job and return job ID.""" # Implementation details... ``` -------------------------------- ### Get Job Status with MCP Background Job Server (Python) Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieves the current status of a background job using its unique job ID. The status can indicate if the job is running, completed successfully, failed, or was killed. This function is useful for polling job progress or determining when a job has finished. ```python from mcp_background_job.server import get_job_status from mcp_background_job.models import JobStatus job_id = "550e8400-e29b-41d4-a716-446655440000" # Check status of a job status = await get_job_status(job_id=job_id) print(f"Job status: {status.status}") # Output: Job status: running # Poll until job completes import asyncio while True: status = await get_job_status(job_id=job_id) if status.status != JobStatus.RUNNING: print(f"Job finished with status: {status.status}") break await asyncio.sleep(2) # Possible status values: # JobStatus.RUNNING - process is executing # JobStatus.COMPLETED - exited with code 0 # JobStatus.FAILED - exited with non-zero code # JobStatus.KILLED - terminated by user # Handle job not found try: status = await get_job_status(job_id="invalid-uuid") except ToolError as e: print(f"Error: {e}") # Job invalid-uuid not found # Returns: StatusOutput(status=JobStatus) ``` -------------------------------- ### Optimize Ring Buffer Sizing Logic Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md This Python snippet addresses potential optimizations for ring buffer sizing. It suggests replacing a rough estimate with a more accurate constant, aiming to improve memory management and efficiency. This is relevant for the `process.py` module. ```python max_lines = max_output_size // 100 # Should be BYTES_PER_LINE_ESTIMATE ``` -------------------------------- ### Python Generic vs. Specific Exception Handling Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md Demonstrates inconsistent exception handling in Python. One instance uses a generic 'except:' which catches all exceptions, while another uses 'except Exception as e:' for more specific error logging and handling. The generic approach can mask errors and make debugging difficult. ```python # service.py:95 - Generic exception handling except: pass # vs service.py:241 - Specific exception handling except Exception as e: logger.warning(f"Failed to update status for job {job_id}: {e}") ``` -------------------------------- ### Interact with Job Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Send input to a running job's stdin and receive immediate output. Useful for interactive processes like REPLs or shells. ```APIDOC ## Interact with Job ### Description Send input to a running job's standard input (stdin) and receive its standard output (stdout) and standard error (stderr). ### Method POST ### Endpoint `/jobs/{job_id}/interact` ### Path Parameters - **job_id** (string) - Required - The unique identifier of the running job to interact with. ### Request Body - **input** (string) - Required - The data to send to the job's stdin. ### Request Example ```json { "input": "print('Hello from MCP')" } ``` ### Response #### Success Response (200) - **stdout** (string) - The standard output from the job. - **stderr** (string) - The standard error from the job. #### Response Example ```json { "stdout": "Hello from MCP\n", "stderr": "" } ``` #### Error Response (400) - **error** (string) - Description of the error (e.g., 'Job is not running'). ``` -------------------------------- ### Programmatic Configuration of Background Job Settings (Python) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This Python code snippet shows how to programmatically configure the mcp-background-job settings using the BackgroundJobConfig class. It allows for setting parameters such as the maximum number of concurrent jobs, the maximum output size in bytes, the default job timeout, and the cleanup interval. ```python from mcp_background_job.config import BackgroundJobConfig config = BackgroundJobConfig( max_concurrent_jobs=20, max_output_size_bytes=20 * 1024 * 1024, # 20MB default_job_timeout=7200, # 2 hours cleanup_interval_seconds=600 # 10 minutes ) ``` -------------------------------- ### Running Tests with pytest (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md These bash commands demonstrate how to execute tests for the mcp-background-job project using pytest. You can run all tests, or specifically target unit or integration tests with the provided options. ```bash # Run all tests uv run pytest tests/ # Run unit tests only uv run pytest tests/unit/ -v # Run integration tests only uv run pytest tests/integration/ -v ``` -------------------------------- ### Code Formatting and Type Checking (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This bash snippet shows commands for maintaining code quality in the mcp-background-job project. It includes formatting the code using ruff and performing type checking with mypy. ```bash # Format code with ruff uv run ruff format # Run type checking uv run mypy src/ ``` -------------------------------- ### Configure mcp-background-job with Environment Variables (JSON) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This JSON configuration illustrates how to set environment variables for the mcp-background-job server within a Claude Code configuration. It specifies the command to run the background job and its arguments, along with key environment variables like max jobs and max output size. ```json { "mcpServers": { "background-job": { "command": "uvx", "args": ["mcp-background-job"], "env": { "MCP_BG_MAX_JOBS": "20", "MCP_BG_MAX_OUTPUT_SIZE": "20MB" } } } } ``` -------------------------------- ### Job Data Models with Pydantic Validation in Python Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Defines Pydantic models for representing job-related data, ensuring type safety and validation. It includes models for complete job information (`BackgroundJob`), job summaries (`JobSummary`), process output (`ProcessOutput`), execution results (`ExecuteOutput`), and kill operations (`KillOutput`). It also demonstrates how to handle validation errors. ```python from mcp_background_job.models import ( BackgroundJob, JobSummary, JobStatus, ProcessOutput, ExecuteOutput, KillOutput ) from datetime import datetime, timezone job = BackgroundJob( job_id="550e8400-e29b-41d4-a716-446655440000", command="npm run dev", status=JobStatus.RUNNING, started=datetime.now(timezone.utc), pid=12345 ) print(f"Job: {job.command} (PID: {job.pid})") job.status = JobStatus.COMPLETED job.completed = datetime.now(timezone.utc) job.exit_code = 0 summary = JobSummary( job_id=job.job_id, status=job.status, command=job.command, started=job.started ) output = ProcessOutput( stdout="Build completed successfully", stderr="" ) print(output.stdout) execute_result = ExecuteOutput(job_id="550e8400-e29b-41d4-a716-446655440000") kill_result = KillOutput(status="killed") try: invalid = BackgroundJob( job_id="not-a-uuid", command="", status="invalid_status" ) except ValidationError as e: print(f"Validation failed: {e}") ``` -------------------------------- ### Test Commands for Background Jobs (Python) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Provides lists of safe and interactive commands suitable for testing background job functionality. These commands cover basic operations, time delays, file system interactions, and interactive shell sessions. ```python # Example test commands SAFE_TEST_COMMANDS = [ "echo 'hello world'", "sleep 5", "python -c 'import time; time.sleep(2); print(\"done\")'", "ls -la", "pwd" ] # Interactive test commands INTERACTIVE_TEST_COMMANDS = [ "python -i", # Python REPL "node", # Node.js REPL "cat" # Read from stdin ] ``` -------------------------------- ### Interact with Job Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Sends input to a running job's standard input and retrieves any immediate output. ```APIDOC ## POST /jobs/{job_id}/interact ### Description Sends input to a job's stdin and returns any immediate output. ### Method POST ### Endpoint /jobs/{job_id}/interact ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier for the job to interact with. #### Request Body - **input** (string) - Required - The input string to send to the job's stdin. ### Request Example ```json { "input": "some input to the job" } ``` ### Response #### Success Response (200) - **stdout** (string) - The standard output from the job after receiving input. - **stderr** (string) - The standard error output from the job after receiving input. #### Response Example ```json { "stdout": "Output from job after receiving input.", "stderr": "" } ``` ``` -------------------------------- ### Logging Debug Messages in Python Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md This Python snippet illustrates a potentially verbose debug log message that may be too detailed for production environments. It highlights the need to review logging levels for optimal performance and resource usage during deployment. ```python logger.debug(f"Process {self.job_id} {stream_name}: {line}") ``` -------------------------------- ### Python Tool Input/Output Models for Job Management Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/SPEC.md Defines Pydantic models for inputs and outputs of various job management operations, such as executing commands, checking status, killing jobs, and retrieving output. These models facilitate structured communication with the background job server. ```python class ExecuteInput(BaseModel): command: str class ExecuteOutput(BaseModel): job_id: str class TailInput(BaseModel): job_id: str lines: int = 50 # Default to last 50 lines class StatusInput(BaseModel): job_id: str class StatusOutput(BaseModel): status: JobStatus class KillInput(BaseModel): job_id: str class KillOutput(BaseModel): status: str # "killed", "already_terminated", "not_found" class OutputInput(BaseModel): job_id: str class InteractInput(BaseModel): job_id: str input: str class ListOutput(BaseModel): jobs: List[JobSummary] ``` -------------------------------- ### Job Status Values Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md Defines the possible status values for background jobs. ```APIDOC ## Job Status Values - `running` - Process is currently executing - `completed` - Process finished successfully - `failed` - Process terminated with error - `killed` - Process was terminated by user ``` -------------------------------- ### JobManager Service Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Manages the lifecycle of background jobs, including execution, tracking, and resource limits. ```APIDOC ## JobManager Service ### Description The JobManager is the core service responsible for creating, monitoring, and managing background jobs. It enforces configuration rules such as maximum concurrent jobs, output size limits, and command restrictions. ### Method POST ### Endpoint `/jobs/execute` ### Request Body - **command** (string) - Required - The command to execute as a background job. - **max_concurrent_jobs** (integer) - Optional - Overrides the default maximum concurrent jobs for this execution. - **max_output_size_bytes** (integer) - Optional - Overrides the default maximum output size for this execution. - **default_job_timeout** (integer) - Optional - Overrides the default job timeout for this execution. ### Request Example ```json { "command": "npm test", "max_concurrent_jobs": 10 } ``` ### Response #### Success Response (200) - **job_id** (string) - The unique identifier of the newly created job. #### Response Example ```json { "job_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef" } ``` #### Error Response (400) - **error** (string) - Description of the error (e.g., 'Command not allowed', 'Max concurrent jobs reached'). ``` -------------------------------- ### List All Background Jobs Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Retrieve a list of all background jobs and their current statuses. This function returns a `ProcessOutput` object containing a list of job summaries. It allows filtering jobs by status and accessing the most recent job. ```python from mcp_background_job.server import list_jobs # Get all jobs jobs = await list_jobs() print(f"Total jobs: {len(jobs.jobs)}") # Display job information for job in jobs.jobs: print(f"Job ID: {job.job_id}") print(f"Status: {job.status}") print(f"Command: {job.command}") print(f"Started: {job.started}") print("---") # Filter by status running_jobs = [j for j in jobs.jobs if j.status == JobStatus.RUNNING] print(f"Running jobs: {len(running_jobs)}") completed = [j for j in jobs.jobs if j.status == JobStatus.COMPLETED] print(f"Completed jobs: {len(completed)}") # Jobs are sorted by start time (newest first) if jobs.jobs: latest_job = jobs.jobs[0] print(f"Most recent job: {latest_job.command}") # Returns: ListOutput(jobs=List[JobSummary]) ``` -------------------------------- ### Interact with Running Job Source: https://context7.com/dylan-gluck/mcp-background-job/llms.txt Send input to a running job's standard input and receive its standard output. This is useful for interactive shells like Python REPLs or custom scripts. It can execute multiple commands sequentially and handle job errors. ```python from mcp_background_job.server import interact_with_job, execute_command, kill_job from mcp_background_job.exceptions import ToolError import asyncio # Start an interactive Python REPL repl = await execute_command(command="python -i") await asyncio.sleep(0.5) # Wait for REPL to start # Send Python code result = await interact_with_job( job_id=repl.job_id, input="print('Hello from MCP')" ) print(result.stdout) # Output: Hello from MCP # Execute multiple commands result = await interact_with_job( job_id=repl.job_id, input="import sys; print(sys.version)" ) print(f"Python version: {result.stdout}") # Interactive shell session shell = await execute_command(command="bash") await asyncio.sleep(0.3) # Run commands in shell result = await interact_with_job(job_id=shell.job_id, input="ls -la") print(result.stdout) result = await interact_with_job(job_id=shell.job_id, input="pwd") print(f"Current directory: {result.stdout}") # Send input to custom script script_job = await execute_command(command="python interactive_script.py") await asyncio.sleep(0.5) response = await interact_with_job( job_id=script_job.job_id, input="user_input_value" ) print(f"Script response: {response.stdout}") # Handle errors try: # Try to interact with completed job result = await interact_with_job( job_id=completed_job_id, # Assume completed_job_id is defined elsewhere input="some input" ) except ToolError as e: print(f"Error: {e}") # Job xxx is not running and cannot accept input # Clean up await kill_job(job_id=repl.job_id) # Returns: ProcessOutput(stdout=str, stderr=str) ``` -------------------------------- ### Configure mcp-background-job with Environment Variables (Bash) Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/README.md This snippet demonstrates how to configure the mcp-background-job server's behavior using environment variables in a bash shell. It covers settings like maximum concurrent jobs, output buffer size, job timeouts, cleanup intervals, working directory, and allowed command patterns. ```bash export MCP_BG_MAX_JOBS=20 export MCP_BG_MAX_OUTPUT_SIZE=20MB # or in bytes: export MCP_BG_MAX_OUTPUT_SIZE=20971520 export MCP_BG_JOB_TIMEOUT=3600 export MCP_BG_CLEANUP_INTERVAL=600 export MCP_BG_WORKING_DIR=/path/to/project export MCP_BG_ALLOWED_COMMANDS="^npm ,^python ,^echo ,^ls" ``` -------------------------------- ### Python: Default Blocked Command Patterns Source: https://github.com/dylan-gluck/mcp-background-job/blob/main/docs/reviews/001_mcp-background-job-server.md Defines a list of default blocked command patterns in Python, intended for use in security configurations. These regular expressions target common dangerous commands like `rm -rf /` and piped shell commands (`wget | bash`). This enhances the robustness of command validation. ```python # In src/mcp_background_job/config.py DEFAULT_BLOCKED_PATTERNS = [ r"rm\s+-rf\s+/", r"wget.*\|.*bash", r"curl.*\|.*sh", # Add more dangerous patterns as needed ] ```