#!/usr/bin/env python3
"""
FastMCP Server for AgentExecMPC.
This module contains the FastMCP server implementation with all MCP tools:
- shell: Execute shell commands with timeout and safety controls
- execute_code: Run code snippets in Python, Node.js, or Go
- install_package: Install packages via pip, npm, or go modules
"""
import asyncio
import os
import signal
import tempfile
from pathlib import Path
from fastmcp import FastMCP
from ..utils import (
MAX_TIMEOUT,
ExecuteCodeRequest,
ExecutionResponse,
InstallPackageRequest,
ShellRequest,
execute_go_code_with_shell,
get_code_executor,
get_package_install_command,
run_subprocess,
signal_handler,
)
# Configuration
MAX_CONCURRENCY = 4
WORKSPACE_DIR = (
Path("/workspace").resolve()
if Path("/workspace").exists()
else Path("./workspace").resolve()
)
# Ensure workspace directory exists
WORKSPACE_DIR.mkdir(exist_ok=True)
# Concurrency semaphore
execution_semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
# Initialize FastMCP server
mcp = FastMCP(
name="AgentExecMPC",
instructions=f"""
AgentExecMPC - Advanced Code Execution and Package Management Server
This MCP server provides comprehensive development and execution capabilities:
SHELL TOOL:
Execute shell commands in a secure sandbox with timeout controls.
Perfect for file operations, system commands, and automation tasks.
CODE EXECUTION:
Run Python, Node.js, and Go code snippets with automatic environment setup.
Supports temporary file creation, execution, and cleanup. All temporary files are deleted after execution. For persistent scripts or files, create them using the SHELL TOOL.
PACKAGE MANAGEMENT:
Install packages via pip (Python), npm (Node.js), or go modules.
Supports version pinning and dependency management.
SECURITY FEATURES:
- Sandboxed workspace execution
- Configurable timeout limits (1-300 seconds)
- Automatic process cleanup
- Input size validation (64KB limit)
- Comprehensive error handling
CONCURRENCY:
- Maximum {MAX_CONCURRENCY} concurrent operations
- Semaphore-controlled execution
- Resource-aware processing
All operations provide detailed execution feedback including stdout, stderr,
exit codes, execution duration, and success status.
""",
)
# MCP Tools
@mcp.tool()
async def shell(request: ShellRequest) -> ExecutionResponse:
"""
Execute shell commands in a secure sandboxed environment.
This tool allows executing arbitrary shell commands with built-in safety controls
including timeout limits, working directory restrictions, and process cleanup.
Perfect for file operations, system commands, and shell scripting tasks.
Features:
- Configurable timeout (default: 60s, max: 300s)
- Custom working directory support
- Automatic process cleanup
- Sandboxed execution in workspace
- Comprehensive error handling
Args:
request: Shell execution parameters including command, timeout, and working directory
Returns:
ExecutionResponse: Complete execution results with stdout, stderr, exit code,
duration, and success status
Examples:
- List files: {"command": "ls -la"}
- Create directory: {"command": "mkdir -p /workspace/new_folder"}
- Run with timeout: {"command": "long_running_task", "timeout": 120}
- Custom directory: {"command": "pwd", "cwd": "/workspace/subfolder"}
"""
async with execution_semaphore:
cwd = Path(request.cwd) if request.cwd else WORKSPACE_DIR
# Ensure the working directory exists and is safe
if not cwd.exists():
return ExecutionResponse(
stdout="",
stderr=f"Working directory does not exist: {cwd}",
exit_code=1,
duration_ms=0,
success=False,
error=f"Working directory does not exist: {cwd}",
)
# Use shell to execute the command
cmd = ["bash", "-c", request.command]
return await run_subprocess(
cmd=cmd, timeout=request.timeout, cwd=cwd, workspace_dir=WORKSPACE_DIR
)
@mcp.tool()
async def execute_code(request: ExecuteCodeRequest) -> ExecutionResponse:
"""
Execute code snippets in Python, Node.js, or Go with automatic environment setup.
This tool creates temporary files and executes code in isolated environments
with proper cleanup and timeout handling. Supports multiple programming
languages with their respective runtimes.
Supported Languages:
- Python: Full Python 3.x environment with standard library
- Node.js/JavaScript: Node.js runtime with npm packages
- Go: Go compiler and runtime environment with CGO_ENABLED=0
Features:
- Automatic temporary file management
- Language-specific execution environments
- Configurable timeout controls
- Detailed execution feedback
- Secure code isolation
- Temporary files are deleted after execution; code is non-persistent. Use the shell tool to create reusable scripts.
Args:
request: Code execution parameters including code, language, and timeout
Returns:
ExecutionResponse: Complete execution results with output, errors,
performance metrics, and success status
Examples:
- Python: {"code": "print('Hello World')", "language": "python"}
- Node.js: {"code": "console.log('Hello World')", "language": "node"}
- Go: {"code": "package main\\nfunc main() { println(\"Hello\") }", "language": "go"}
"""
async with execution_semaphore:
try:
# Special handling for Go code execution
if request.language.lower() == "go":
return await execute_go_code_with_shell(
request.code, request.timeout, WORKSPACE_DIR
)
# Get the appropriate executor for other languages
executor_cmd = get_code_executor(request.language)
# Create temporary file for the code
suffix_map = {
"python": ".py",
"node": ".js",
"javascript": ".js",
"js": ".js",
"go": ".go",
}
suffix = suffix_map.get(request.language, ".txt")
with tempfile.NamedTemporaryFile(
mode="w", suffix=suffix, dir=WORKSPACE_DIR, delete=False
) as temp_file:
temp_file.write(request.code)
temp_file_path = temp_file.name
try:
# Prepare command
cmd = executor_cmd + [temp_file_path]
# Execute the code
result = await run_subprocess(
cmd=cmd, timeout=request.timeout, workspace_dir=WORKSPACE_DIR
)
return result
finally:
# Clean up temporary file
try:
os.unlink(temp_file_path)
except OSError:
pass # File might already be deleted
except Exception as e:
return ExecutionResponse(
stdout="",
stderr=str(e),
exit_code=1,
duration_ms=0,
success=False,
error=f"Code execution setup failed: {str(e)}",
)
@mcp.tool()
async def install_package(request: InstallPackageRequest) -> ExecutionResponse:
"""
Install packages using pip, npm, or Go modules with version control.
This tool provides a unified interface for package installation across
different programming environments. Supports version pinning and provides
detailed installation feedback with proper error handling.
Supported Package Managers:
- pip: Python package installer (PyPI packages)
- npm: Node.js package manager (npm registry)
- go: Go module system (Go packages)
Features:
- Version pinning support
- Installation progress tracking
- Dependency resolution
- Extended timeout for large packages
- Comprehensive error reporting
Args:
request: Package installation parameters including name, manager, and version
Returns:
ExecutionResponse: Installation results with output, any errors,
and installation success status
Examples:
- Install Python package: {"package": "requests", "package_manager": "pip"}
- Install with version: {"package": "lodash", "package_manager": "npm", "version": "4.17.21"}
- Install Go module: {"package": "github.com/gin-gonic/gin", "package_manager": "go"}
"""
async with execution_semaphore:
try:
# Get the installation command
cmd = get_package_install_command(
request.package_manager, request.package, request.version
)
# Execute the installation
result = await run_subprocess(
cmd=cmd,
timeout=300, # Package installations can take longer
workspace_dir=WORKSPACE_DIR,
)
return result
except Exception as e:
return ExecutionResponse(
stdout="",
stderr=str(e),
exit_code=1,
duration_ms=0,
success=False,
error=f"Package installation failed: {str(e)}",
)
# Register signal handlers
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
def run_server():
"""Run the FastMCP server with configurable transport."""
# Check environment for transport type
transport = os.environ.get("MCP_TRANSPORT", "stdio").lower()
port = int(os.environ.get("MCP_PORT", "8000"))
host = os.environ.get("MCP_HOST", "0.0.0.0")
path = os.environ.get("MCP_PATH", "/mcp")
print("Starting AgentExecMPC server...")
print(f"Workspace directory: {WORKSPACE_DIR}")
print(f"Max timeout: {MAX_TIMEOUT}s")
print(f"Max concurrency: {MAX_CONCURRENCY}")
print(f"Transport: {transport}")
if transport == "sse":
print(f"SSE server listening on {host}:{port}")
# Run the FastMCP server with SSE transport (legacy)
mcp.run(transport="sse", host=host, port=port)
elif transport == "streamable-http":
print(f"Streamable HTTP server listening on {host}:{port}{path}")
# Run the FastMCP server with Streamable HTTP transport (recommended)
mcp.run(transport="streamable-http", host=host, port=port, path=path)
else:
print("Using stdio transport")
# Run the FastMCP server with stdio transport
mcp.run(transport="stdio")
if __name__ == "__main__":
run_server()
