server

server

Marianne MCP Server implementation.

This module implements the core MCP server that exposes Marianne job management capabilities through the Model Context Protocol. The server provides tools and resources for external AI agents to interact with Marianne.

The server implements JSON-RPC 2.0 over HTTP/SSE transport and follows the MCP specification for capability negotiation, tool execution, and resource access.

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