"""
Pydantic models for the AgentExecMPC server.
This module contains all the request and response models used by the MCP tools.
"""
import re
from pydantic import BaseModel, Field, field_validator
# Configuration constants
DEFAULT_TIMEOUT = 60 # seconds
MAX_TIMEOUT = 300 # seconds
MAX_INPUT_SIZE = 64 * 1024 # 64 KB
class ShellRequest(BaseModel):
command: str = Field(
...,
description="Shell command to execute. Can be any valid bash command including pipes, redirects, and multi-line scripts. Maximum size: 64KB",
examples=["ls -la", "find . -name '*.py'", "echo 'Hello' | grep Hello"],
)
timeout: int | None = Field(
DEFAULT_TIMEOUT,
description=f"Maximum execution time in seconds. Range: 1-{MAX_TIMEOUT}. Default: {DEFAULT_TIMEOUT}s",
ge=1,
le=MAX_TIMEOUT,
)
cwd: str | None = Field(
None,
description="Working directory path for command execution. Must be an existing directory. Defaults to the secure workspace directory if not specified",
examples=["/workspace", "/workspace/project", "./subdirectory"],
)
@field_validator("timeout")
def validate_timeout(cls, v):
if v and v > MAX_TIMEOUT:
raise ValueError(f"Timeout cannot exceed {MAX_TIMEOUT} seconds")
return v or DEFAULT_TIMEOUT
@field_validator("command")
def validate_command_size(cls, v):
if len(v.encode("utf-8")) > MAX_INPUT_SIZE:
raise ValueError(f"Command size cannot exceed {MAX_INPUT_SIZE} bytes")
return v
class ExecuteCodeRequest(BaseModel):
code: str = Field(
...,
description="Source code to execute. Must be valid syntax for the specified language. Maximum size: 64KB",
examples=[
"print('Hello World')",
"console.log('Hello from Node.js')",
'package main\n\nfunc main() {\n println("Hello Go")\n}',
],
)
language: str = Field(
...,
description="Programming language for code execution. Supported: 'python' (Python 3.x), 'node'/'javascript'/'js' (Node.js), 'go' (Go compiler)",
examples=["python", "node", "javascript", "go"],
)
timeout: int | None = Field(
DEFAULT_TIMEOUT,
description=f"Maximum execution time in seconds. Range: 1-{MAX_TIMEOUT}. Default: {DEFAULT_TIMEOUT}s. Recommended: 30s for simple scripts, 60s+ for complex operations",
ge=1,
le=MAX_TIMEOUT,
)
@field_validator("timeout")
def validate_timeout(cls, v):
if v and v > MAX_TIMEOUT:
raise ValueError(f"Timeout cannot exceed {MAX_TIMEOUT} seconds")
return v or DEFAULT_TIMEOUT
@field_validator("code")
def validate_code_size(cls, v):
if len(v.encode("utf-8")) > MAX_INPUT_SIZE:
raise ValueError(f"Code size cannot exceed {MAX_INPUT_SIZE} bytes")
return v
@field_validator("language")
def validate_language(cls, v):
supported = {"python", "node", "go", "javascript", "js"}
if v.lower() not in supported:
raise ValueError(f"Unsupported language: {v}. Supported: {supported}")
return v.lower()
class InstallPackageRequest(BaseModel):
package: str = Field(
...,
description="Package name to install. Must follow the naming conventions of the respective package manager (e.g., 'requests' for pip, 'lodash' for npm, 'github.com/gin-gonic/gin' for go)",
examples=["requests", "numpy", "lodash", "express", "github.com/gin-gonic/gin"],
)
package_manager: str = Field(
...,
description="Package manager to use. 'pip' for Python packages, 'npm' for Node.js packages, 'go' for Go modules",
examples=["pip", "npm", "go"],
)
version: str | None = Field(
None,
description="Specific package version to install. If not specified, installs the latest version. Format varies by package manager (e.g., '2.28.1', '^4.17.0', 'v1.8.0')",
examples=["2.28.1", "^4.17.0", "latest", "v1.8.0"],
)
@field_validator("package_manager")
def validate_package_manager(cls, v):
supported = {"pip", "npm", "go"}
if v.lower() not in supported:
raise ValueError(
f"Unsupported package manager: {v}. Supported: {supported}"
)
return v.lower()
@field_validator("package")
def validate_package_name(cls, v):
# Basic package name validation
if not re.match(r"^[a-zA-Z0-9\-_./@]+$", v):
raise ValueError("Invalid package name format")
return v
class ExecutionResponse(BaseModel):
stdout: str = Field(
..., description="Standard output from the executed command/code"
)
stderr: str = Field(
..., description="Standard error output from the executed command/code"
)
exit_code: int = Field(
...,
description="Process exit code (0 indicates success, non-zero indicates error)",
)
duration_ms: int = Field(
..., description="Total execution duration in milliseconds"
)
success: bool = Field(
..., description="Whether the execution completed successfully (exit_code == 0)"
)
error: str | None = Field(
None,
description="Human-readable error message if execution failed or encountered issues",
)
