Axom MCP Server

Instructions

Implementation Reference

• src/axom_mcp/handlers/exec.py:57-82 (handler)

  Main handler function handle_exec that routes axom_mcp_exec tool calls to appropriate operation handlers (read, write, shell) with input validation using ExecInput schema
  async def handle_exec(arguments: Dict[str, Any]) -> str: """Handle axom_mcp_exec tool calls. Args: arguments: Tool arguments containing operation and parameters Returns: JSON string with operation result """ # Validate input input_data = ExecInput(**arguments) operation = input_data.operation target = input_data.target try: if operation == "read": return await _handle_read(target) elif operation == "write": return await _handle_write(target, input_data.data) elif operation == "shell": return await _handle_shell(target) else: return json.dumps({"error": f"Unknown operation: {operation}"}) except Exception as e: logger.error(f"Exec operation failed: {e}") return json.dumps({"error": str(e)})
• src/axom_mcp/handlers/exec.py:85-195 (handler)

  Operation-specific handler functions: _handle_read (lines 85-116), _handle_write (lines 119-150), and _handle_shell (lines 153-195) that implement file operations and shell command execution with security validations
  async def _handle_read(target: str) -> str: """Read file contents.""" try: path = _validate_path(target) if not path.exists(): return json.dumps({"error": f"File not found: {target}"}) if not path.is_file(): return json.dumps({"error": f"Not a file: {target}"}) # Check file size if path.stat().st_size > MAX_FILE_SIZE: return json.dumps({ "error": f"File too large: {target} (max {MAX_FILE_SIZE} bytes)" }) # Read file content content = path.read_text(encoding="utf-8", errors="replace") return json.dumps({ "success": True, "operation": "read", "target": str(path), "content": content, "size": len(content), }) except ValueError as e: return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Failed to read file: {e}") return json.dumps({"error": str(e)}) async def _handle_write(target: str, data: Optional[str]) -> str: """Write data to file.""" # Check if write operations are allowed (AXOM_READ_ONLY defaults to False) if _env_flag_enabled("AXOM_READ_ONLY", default=False): return json.dumps({ "error": "Write operations disabled. AXOM_READ_ONLY is enabled." }) if data is None: return json.dumps({"error": "data is required for write operation"}) try: path = _validate_path(target) # Create parent directories if needed path.parent.mkdir(parents=True, exist_ok=True) # Write file path.write_text(data, encoding="utf-8") return json.dumps({ "success": True, "operation": "write", "target": str(path), "size": len(data), "message": f"Successfully wrote {len(data)} bytes to {path}", }) except ValueError as e: return json.dumps({"error": str(e)}) except Exception as e: logger.error(f"Failed to write file: {e}") return json.dumps({"error": str(e)}) async def _handle_shell(command: str) -> str: """Execute shell command.""" # Check if shell operations are allowed (AXOM_READ_ONLY defaults to False) if _env_flag_enabled("AXOM_READ_ONLY", default=False): return json.dumps({ "error": "Shell operations disabled. AXOM_READ_ONLY is enabled." }) try: # Execute command with timeout process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) try: stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=60.0 # 60 second timeout ) except asyncio.TimeoutError: process.kill() return json.dumps({ "error": "Command timed out after 60 seconds", "command": command, }) # Decode output stdout_str = stdout.decode("utf-8", errors="replace") if stdout else "" stderr_str = stderr.decode("utf-8", errors="replace") if stderr else "" return json.dumps({ "success": process.returncode == 0, "operation": "shell", "command": command, "exit_code": process.returncode, "stdout": stdout_str, "stderr": stderr_str, }) except Exception as e: logger.error(f"Failed to execute shell command: {e}") return json.dumps({"error": str(e)})
• src/axom_mcp/schemas.py:209-232 (schema)

  ExecInput Pydantic schema defining the input validation for axom_mcp_exec tool with operation (read/write/shell), target (file path or command), data (for writes), and optional chain parameter
  class ExecInput(BaseModel): """Input schema for axom_mcp_exec tool.""" model_config = {"extra": "forbid"} operation: str = Field( ..., pattern="^(read|write|shell)$", description="Operation type: read, write, or shell", ) target: str = Field( ..., min_length=1, max_length=4096, description="File path for read/write, or command for shell", ) data: Optional[str] = Field( default=None, max_length=10_000_000, description="Data to write (for write operation)", ) chain: Optional[List[Dict[str, Any]]] = Field( default=None, max_length=10, description="Chain of subsequent operations" )
• src/axom_mcp/server.py:171-221 (registration)

  Tool registration in TOOLS list defining axom_mcp_exec with name, description, inputSchema, and annotations for MCP protocol
  name="axom_mcp_exec", description="""Execute file operations and shell commands with chain-reaction support. Operations: - read: Read file contents from allowed directories - write: Write data to files (unless AXOM_READ_ONLY=true) - shell: Execute shell commands (unless AXOM_READ_ONLY=true) Chain Reactions: Chain multiple operations together using the chain parameter. Each step can reference the previous result using ${_result} variable substitution. Example: { "operation": "read", "target": "/file.txt", "chain": [ { "tool": "axom_mcp_transform", "args": {"input": "${_result.content}", "output_format": "json"} } ] } Security: - File operations restricted to allowed directories (cwd, ~/) - Shell/write operations enabled by default (set AXOM_READ_ONLY=true to disable) - Input size limits: 10MB max for files""", inputSchema={ "type": "object", "properties": { "operation": { "type": "string", "enum": ["read", "write", "shell"], "description": "Operation type", }, "target": {"type": "string", "description": "File path or command"}, "data": { "type": "string", "description": "Data to write (for write operation)", }, "chain": { "type": "array", "items": {"type": "object"}, "description": "Chain of subsequent operations", }, }, "required": ["operation", "target"], }, annotations=TOOL_ANNOTATIONS["exec"], ),
• src/axom_mcp/server.py:486-506 (registration)

  call_tool function that routes incoming tool requests, specifically line 492-493 where axom_mcp_exec calls are forwarded to handle_exec handler
  @server.call_tool() async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: """Handle tool calls.""" try: if name == "axom_mcp_memory": result = await handle_memory(arguments) elif name == "axom_mcp_exec": result = await handle_exec(arguments) elif name == "axom_mcp_analyze": result = await handle_analyze(arguments) elif name == "axom_mcp_discover": result = await handle_discover(arguments) elif name == "axom_mcp_transform": result = await handle_transform(arguments) else: return [TextContent(type="text", text=f"Unknown tool: {name}")] return [TextContent(type="text", text=result)] except Exception as e: logger.error(f"Tool call failed: {name} - {e}") return [TextContent(type="text", text=f"Error: {str(e)}")]