Index

mcp

Marianne MCP Server - Model Context Protocol integration.

This module implements an MCP server that exposes Marianne's job management capabilities as tools for external AI agents. The server provides:

• Job lifecycle tools (run, pause, resume, cancel)
• Status monitoring and log streaming
• Artifact management and workspace browsing
• Mzt configuration as resources

Example

from marianne.mcp.server import MCPServer server = MCPServer() await server.serve()

Classes

MCPServer

MCPServer(workspace_root=None)

Marianne MCP Server - Exposes Marianne capabilities via Model Context Protocol.

The server implements the MCP specification to provide: - Job management tools (run, pause, resume, cancel, status) - Artifact browsing and log streaming - Configuration access as resources

Security: - All tool executions require explicit user consent - File system access is restricted to Marianne workspace directories - No arbitrary code execution beyond Marianne's built-in capabilities

Attributes:

Name	Type	Description
tools		Available MCP tools grouped by category
resources		Available MCP resources
capabilities	dict[str, Any]	Server capabilities advertised during negotiation

Example

server = MCPServer() await server.initialize()

Server is ready to accept MCP connections

Initialize the MCP server.

Parameters:

Name	Type	Description	Default
workspace_root	Path | None	Optional root directory for workspace operations. Defaults to current working directory.	None

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

def __init__(self, workspace_root: Path | None = None):
    """Initialize the MCP server.

    Args:
        workspace_root: Optional root directory for workspace operations.
                       Defaults to current working directory.
    """
    self.workspace_root = workspace_root or Path.cwd()
    self.state_backend = JsonStateBackend(self.workspace_root)

    # Initialize tool categories
    self.job_tools = JobTools(self.state_backend, self.workspace_root)
    self.control_tools = ControlTools(self.state_backend, self.workspace_root)
    self.artifact_tools = ArtifactTools(self.workspace_root)
    self.score_tools = ScoreTools(self.workspace_root)

    # Initialize resources
    self.config_resources = ConfigResources(self.state_backend, self.workspace_root)

    # Server state
    self.initialized = False
    self.client_info: dict[str, Any] | None = None

Attributes

capabilities property

capabilities

Server capabilities advertised during MCP negotiation.

Functions

initialize async

initialize(client_info=None)

Initialize the server with client capabilities.

Parameters:

Name	Type	Description	Default
client_info	dict[str, Any] | None	Information about the connecting MCP client	None

Returns:

Type	Description
dict[str, Any]	Server initialization response with capabilities

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

async def initialize(self, client_info: dict[str, Any] | None = None) -> dict[str, Any]:
    """Initialize the server with client capabilities.

    Args:
        client_info: Information about the connecting MCP client

    Returns:
        Server initialization response with capabilities
    """
    self.client_info = client_info or {}
    self.initialized = True

    client_name = self.client_info.get('name', 'unknown')
    logger.info(f"MCP Server initialized with client: {client_name}")

    return {
        "capabilities": self.capabilities,
        "serverInfo": {
            "name": "marianne-mcp-server",
            "version": "1.0.0",
            "description": "Marianne AI Compose MCP Server - Job management and orchestration"
        }
    }

list_tools async

list_tools()

List all available tools.

Returns:

Type	Description
list[dict[str, Any]]	List of tool definitions with schemas

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

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

    Returns:
        List of tool definitions with schemas
    """
    if not self.initialized:
        raise RuntimeError("Server not initialized")

    tools = []
    tools.extend(await self.job_tools.list_tools())
    tools.extend(await self.control_tools.list_tools())
    tools.extend(await self.artifact_tools.list_tools())
    tools.extend(await self.score_tools.list_tools())

    return tools

call_tool async

call_tool(name, arguments=None)

Execute a tool with the given arguments.

Parameters:

Name	Type	Description	Default
name	str	Tool name to execute	required
arguments	dict[str, Any] | None	Tool arguments	None

Returns:

Type	Description
dict[str, Any]	Tool execution result

Raises:

Type	Description
ValueError	If tool is not found or arguments are invalid
RuntimeError	If server is not initialized

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

async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]:
    """Execute a tool with the given arguments.

    Args:
        name: Tool name to execute
        arguments: Tool arguments

    Returns:
        Tool execution result

    Raises:
        ValueError: If tool is not found or arguments are invalid
        RuntimeError: If server is not initialized
    """
    if not self.initialized:
        raise RuntimeError("Server not initialized")

    arguments = arguments or {}

    # Route to appropriate tool handler
    job_tool_names = ["list_jobs", "get_job", "start_job"]
    control_tool_names = ["pause_job", "resume_job", "cancel_job"]
    artifact_tool_names = [
        "marianne_artifact_list", "marianne_artifact_read",
        "marianne_artifact_get_logs",
        "marianne_artifact_list_artifacts",
        "marianne_artifact_get_artifact",
    ]
    score_tool_names = ["validate_score", "generate_score"]

    if name in job_tool_names:
        return await self.job_tools.call_tool(name, arguments)
    elif name in control_tool_names:
        return await self.control_tools.call_tool(name, arguments)
    elif name in artifact_tool_names:
        return await self.artifact_tools.call_tool(name, arguments)
    elif name in score_tool_names:
        return await self.score_tools.call_tool(name, arguments)
    else:
        raise ValueError(f"Unknown tool: {name}")

list_resources async

list_resources()

List all available resources.

Returns:

Type	Description
list[dict[str, Any]]	List of resource definitions

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

async def list_resources(self) -> list[dict[str, Any]]:
    """List all available resources.

    Returns:
        List of resource definitions
    """
    if not self.initialized:
        raise RuntimeError("Server not initialized")

    resources = []
    resources.extend(await self.config_resources.list_resources())

    return resources

read_resource async

read_resource(uri)

Read a resource by URI.

Parameters:

Name	Type	Description	Default
uri	str	Resource URI to read	required

Returns:

Type	Description
dict[str, Any]	Resource content

Raises:

Type	Description
ValueError	If resource URI is not found

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

async def read_resource(self, uri: str) -> dict[str, Any]:
    """Read a resource by URI.

    Args:
        uri: Resource URI to read

    Returns:
        Resource content

    Raises:
        ValueError: If resource URI is not found
    """
    if not self.initialized:
        raise RuntimeError("Server not initialized")

    # Route to appropriate resource handler
    if uri.startswith("config://") or uri.startswith("marianne://"):
        return await self.config_resources.read_resource(uri)
    else:
        raise ValueError(f"Unknown resource URI: {uri}")

shutdown async

shutdown()

Shutdown the server and cleanup resources.

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

async def shutdown(self) -> None:
    """Shutdown the server and cleanup resources."""
    logger.info("MCP Server shutting down")

    # Cleanup any running operations
    await self.job_tools.shutdown()
    await self.control_tools.shutdown()
    await self.artifact_tools.shutdown()
    await self.score_tools.shutdown()

    self.initialized = False