Skip to content

resources

resources

Marianne MCP Resources - Resource implementations for Mzt configuration access.

This module implements MCP resources that expose Mzt configuration and documentation as readable content. Resources provide context and reference material for AI agents working with Marianne.

Resources are organized by category: - ConfigResources: Access to Mzt configuration schemas and examples

Attributes

Classes

ConfigResources 

ConfigResources(state_backend=None, workspace_root=None)

Mzt configuration resources.

Provides access to Mzt configuration schemas, examples, and documentation as MCP resources. These resources help AI agents understand Marianne's configuration format and available options.

Source code in src/marianne/mcp/resources.py 
35
36
37
38
39
40
41
def __init__(
    self, state_backend: StateBackend | None = None, workspace_root: Path | None = None
) -> None:
    # Base project directory (assuming we're in src/marianne/mcp/)
    self.project_root = Path(__file__).parent.parent.parent.parent
    self.state_backend = state_backend
    self.workspace_root = workspace_root or Path.cwd()
Functions
list_resources async 
list_resources()

List all configuration resources.

Source code in src/marianne/mcp/resources.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def list_resources(self) -> list[dict[str, Any]]:
    """List all configuration resources."""
    resources = [
        {
            "uri": "config://schema",
            "name": "Mzt configuration Schema",
            "description": "Complete JSON schema for Marianne job configuration files",
            "mimeType": _CONTENT_TYPE_JSON
        },
        {
            "uri": "config://example",
            "name": "Mzt configuration Example",
            "description": "Example Marianne job configuration with common patterns",
            "mimeType": "text/yaml"
        },
        {
            "uri": "config://backend-options",
            "name": "Backend Configuration Options",
            "description": "Available backend types and their configuration options",
            "mimeType": _CONTENT_TYPE_JSON
        },
        {
            "uri": "config://validation-types",
            "name": "Validation Types Reference",
            "description": "Available validation types and their parameters",
            "mimeType": _CONTENT_TYPE_JSON
        },
        {
            "uri": "config://learning-options",
            "name": "Learning Configuration Options",
            "description": "Learning system configuration parameters and patterns",
            "mimeType": _CONTENT_TYPE_JSON
        },
        # Job management resources
        {
            "uri": "marianne://jobs",
            "name": "Marianne Jobs Overview",
            "description": "List of all Marianne jobs with status and metadata",
            "mimeType": _CONTENT_TYPE_JSON
        },
        {
            "uri": "marianne://templates",
            "name": "Marianne Job Templates",
            "description": "Collection of Marianne job configuration templates",
            "mimeType": _CONTENT_TYPE_JSON
        }
    ]

    # Dynamic job detail resources - only available if we have state backend
    if self.state_backend:
        resources.append({
            "uri": "marianne://jobs/{job_id}",
            "name": "Marianne Job Details (Template)",
            "description": "Detailed information about a specific Marianne job",
            "mimeType": _CONTENT_TYPE_JSON
        })

    return resources
read_resource async 
read_resource(uri)

Read a configuration resource by URI.

Source code in src/marianne/mcp/resources.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def read_resource(self, uri: str) -> dict[str, Any]:
    """Read a configuration resource by URI."""
    try:
        handler_name = self._URI_HANDLERS.get(uri)
        if handler_name:
            result: dict[str, Any] = await getattr(self, handler_name)()
            return result

        if uri.startswith("marianne://jobs/"):
            job_id = uri.replace("marianne://jobs/", "")
            return await self._get_job_details(job_id)

        raise ValueError(f"Unknown resource URI: {uri}")

    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as e:
        logger.warning("Error reading resource %s: %s", uri, e)
        return {
            "contents": [
                {
                    "uri": uri,
                    "mimeType": "text/plain",
                    "text": f"Error reading resource: {str(e)}"
                }
            ]
        }