tools

tools

Marianne MCP Tools - Tool implementations for Marianne job management.

This module implements MCP tools that expose Marianne's job management capabilities to external AI agents. Tools are organized by category:

• JobTools: Job lifecycle management (list, get, start)
• ControlTools: Job control operations (pause, resume, cancel)
• ArtifactTools: Workspace and artifact management

Each tool follows the MCP specification for parameter schemas and return values. Tools leverage the existing JobControlService for consistent behavior with the dashboard.

Classes

JobTools

JobTools(state_backend, workspace_root)

Marianne job lifecycle management tools.

Provides MCP tools for running, monitoring, and querying Marianne jobs. Tools require explicit user consent due to file system and process execution.

Routes through the conductor for all operations.

Source code in src/marianne/mcp/tools.py

def __init__(self, state_backend: JsonStateBackend, workspace_root: Path):
    self.state_backend = state_backend
    self._daemon_client = DaemonClient(_resolve_socket_path(None))
    self.job_control = JobControlService(self._daemon_client)

Functions

list_tools async

list_tools()

List all job management tools.

Source code in src/marianne/mcp/tools.py

async def list_tools(self) -> list[dict[str, Any]]:
    """List all job management tools."""
    return [
        {
            "name": "list_jobs",
            "description": "List all Marianne jobs with their current status",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "status_filter": {
                        "type": "string",
                        "description": (
                            "Filter jobs by status"
                            " (running, paused, completed, failed, cancelled)"
                        ),
                        "enum": ["running", "paused", "completed", "failed", "cancelled"],
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum number of jobs to return",
                        "default": 50,
                        "minimum": 1,
                        "maximum": 500,
                    },
                },
                "required": [],
            },
        },
        {
            "name": "get_job",
            "description": "Get detailed information about a specific Marianne job",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string", "description": "Marianne job ID to retrieve"}
                },
                "required": ["job_id"],
            },
        },
        {
            "name": "start_job",
            "description": "Start a new Marianne job from a configuration file",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "config_path": {
                        "type": "string",
                        "description": "Path to the Marianne job configuration file (.yaml)",
                    },
                    "workspace": {
                        "type": "string",
                        "description": "Workspace directory for job execution (optional)",
                    },
                    "start_sheet": {
                        "type": "integer",
                        "description": "Sheet number to start from (1-indexed)",
                        "default": 1,
                        "minimum": 1,
                    },
                    "self_healing": {
                        "type": "boolean",
                        "description": "Enable self-healing mode for automatic error recovery",
                        "default": False,
                    },
                },
                "required": ["config_path"],
            },
        },
    ]

call_tool async

call_tool(name, arguments)

Execute a job management tool.

Source code in src/marianne/mcp/tools.py

async def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """Execute a job management tool."""
    try:
        if name == "list_jobs":
            return await self._list_jobs(arguments)
        elif name == "get_job":
            return await self._get_job(arguments)
        elif name == "start_job":
            return await self._start_job(arguments)
        else:
            raise ValueError(f"Unknown job tool: {name}")

    except (
        KeyError,
        ValueError,
        FileNotFoundError,
        RuntimeError,
        OSError,
        ConnectionError,
    ) as e:
        logger.exception("Error executing tool %s", name)
        return _make_error_response(e)

shutdown async

shutdown()

Cleanup job tools.

Source code in src/marianne/mcp/tools.py

async def shutdown(self) -> None:
    """Cleanup job tools."""
    pass  # No persistent resources to cleanup

ControlTools

ControlTools(state_backend, workspace_root)

Marianne job control tools.

Provides MCP tools for controlling running Marianne jobs (pause, resume, cancel). These tools interact with job processes and require user consent.

Routes through the conductor for all operations.

Source code in src/marianne/mcp/tools.py

def __init__(self, state_backend: JsonStateBackend, workspace_root: Path):
    self.state_backend = state_backend
    daemon_client = DaemonClient(_resolve_socket_path(None))
    self.job_control = JobControlService(daemon_client)

Functions

list_tools async

list_tools()

List all job control tools.

Source code in src/marianne/mcp/tools.py

async def list_tools(self) -> list[dict[str, Any]]:
    """List all job control tools."""
    return [
        {
            "name": "pause_job",
            "description": "Pause a running Marianne job gracefully at sheet boundary",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string", "description": "Marianne job ID to pause"}
                },
                "required": ["job_id"],
            },
        },
        {
            "name": "resume_job",
            "description": "Resume a paused Marianne job",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string", "description": "Marianne job ID to resume"}
                },
                "required": ["job_id"],
            },
        },
        {
            "name": "cancel_job",
            "description": "Cancel a running Marianne job permanently",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string", "description": "Marianne job ID to cancel"}
                },
                "required": ["job_id"],
            },
        },
    ]

call_tool async

call_tool(name, arguments)

Execute a job control tool.

Source code in src/marianne/mcp/tools.py

async def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """Execute a job control tool."""
    try:
        if name == "pause_job":
            return await self._pause_job(arguments)
        elif name == "resume_job":
            return await self._resume_job(arguments)
        elif name == "cancel_job":
            return await self._cancel_job(arguments)
        else:
            raise ValueError(f"Unknown control tool: {name}")

    except (KeyError, ValueError, RuntimeError, OSError, ConnectionError) as e:
        logger.exception("Error executing control tool %s", name)
        return _make_error_response(e)

shutdown async

shutdown()

Cleanup control tools.

Source code in src/marianne/mcp/tools.py

async def shutdown(self) -> None:
    """Cleanup control tools."""
    pass

ArtifactTools

ArtifactTools(workspace_root)

Marianne artifact and workspace management tools.

Provides MCP tools for browsing workspace files and accessing job artifacts. File system access is restricted to designated workspace directories.

Source code in src/marianne/mcp/tools.py

def __init__(self, workspace_root: Path):
    self.workspace_root = workspace_root
    self._custom_level_cache: dict[str, re.Pattern[str]] = {}

Functions

list_tools async

list_tools()

List all artifact management tools.

Source code in src/marianne/mcp/tools.py

async def list_tools(self) -> list[dict[str, Any]]:
    """List all artifact management tools."""
    return list(_ARTIFACT_TOOL_SCHEMAS)

call_tool async

call_tool(name, arguments)

Execute an artifact management tool.

Source code in src/marianne/mcp/tools.py

async def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """Execute an artifact management tool."""
    dispatch = {
        "marianne_artifact_list": self._list_files,
        "marianne_artifact_read": self._read_file,
        "marianne_artifact_get_logs": self._get_logs,
        "marianne_artifact_list_artifacts": self._list_artifacts,
        "marianne_artifact_get_artifact": self._get_artifact,
    }
    handler = dispatch.get(name)
    if handler is None:
        return {
            "content": [{"type": "text", "text": f"Error: Unknown artifact tool: {name}"}],
            "isError": True,
        }
    try:
        return await handler(arguments)
    except (
        KeyError,
        ValueError,
        FileNotFoundError,
        IsADirectoryError,
        NotADirectoryError,
        PermissionError,
        OSError,
    ) as e:
        logger.exception("Error executing artifact tool %s", name)
        return _make_error_response(e)

shutdown async

shutdown()

Cleanup artifact tools.

Source code in src/marianne/mcp/tools.py

async def shutdown(self) -> None:
    """Cleanup artifact tools."""
    pass

ScoreTools

ScoreTools(workspace_root)

Marianne code quality score tools.

Provides MCP tools for validating and generating quality scores for code changes using Marianne's AI-powered review system. Tools analyze git diffs and provide detailed feedback on code quality, test coverage, security, and documentation.

Source code in src/marianne/mcp/tools.py

def __init__(self, workspace_root: Path):
    self.workspace_root = workspace_root

Functions

list_tools async

list_tools()

List all score management tools.

Returns empty list because validate_score and generate_score are stub implementations. Registering stubs misleads MCP clients into expecting working functionality. Re-enable when backend integration is complete.

Source code in src/marianne/mcp/tools.py

async def list_tools(self) -> list[dict[str, Any]]:
    """List all score management tools.

    Returns empty list because validate_score and generate_score are stub
    implementations. Registering stubs misleads MCP clients into expecting
    working functionality. Re-enable when backend integration is complete.
    """
    return []

call_tool async

call_tool(name, arguments)

Execute a score tool.

Source code in src/marianne/mcp/tools.py

async def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    """Execute a score tool."""
    try:
        if name == "validate_score":
            return await self._validate_score(arguments)
        elif name == "generate_score":
            return await self._generate_score(arguments)
        else:
            raise ValueError(f"Unknown score tool: {name}")

    except (KeyError, ValueError, FileNotFoundError, PermissionError, OSError) as e:
        logger.exception("Error executing score tool %s", name)
        return _make_error_response(e)

shutdown async

shutdown()

Cleanup score tools.

Source code in src/marianne/mcp/tools.py

async def shutdown(self) -> None:
    """Cleanup score tools."""
    pass