From 1c863f0f0c6de1262260f01e0e3e4d9ce55121d8 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Thu, 12 Mar 2026 15:31:10 -0400 Subject: [PATCH] feat(models): Add MCP configuration models and loading logic --- src/models.py | 55 ++++++++++++++++++++++++++++++++++++++++ tests/test_mcp_config.py | 53 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 108 insertions(+) create mode 100644 tests/test_mcp_config.py diff --git a/src/models.py b/src/models.py index ec13288..f10c259 100644 --- a/src/models.py +++ b/src/models.py @@ -37,6 +37,8 @@ See Also: - src/project_manager.py for persistence layer """ from __future__ import annotations +import json +import os import tomllib import datetime from dataclasses import dataclass, field @@ -515,3 +517,56 @@ class Persona: bias_profile=data.get("bias_profile"), ) +@dataclass +class MCPServerConfig: + name: str + command: Optional[str] = None + args: List[str] = field(default_factory=list) + url: Optional[str] = None + auto_start: bool = False + + def to_dict(self) -> Dict[str, Any]: + res = {'auto_start': self.auto_start} + if self.command: res['command'] = self.command + if self.args: res['args'] = self.args + if self.url: res['url'] = self.url + return res + + @classmethod + def from_dict(cls, name: str, data: Dict[str, Any]) -> 'MCPServerConfig': + return cls( + name=name, + command=data.get('command'), + args=data.get('args', []), + url=data.get('url'), + auto_start=data.get('auto_start', False), + ) + +@dataclass +class MCPConfiguration: + mcpServers: Dict[str, MCPServerConfig] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + 'mcpServers': {name: cfg.to_dict() for name, cfg in self.mcpServers.items()} + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'MCPConfiguration': + raw_servers = data.get('mcpServers', {}) + parsed_servers = { + name: MCPServerConfig.from_dict(name, cfg) + for name, cfg in raw_servers.items() + } + return cls(mcpServers=parsed_servers) + +def load_mcp_config(path: str) -> MCPConfiguration: + if not os.path.exists(path): + return MCPConfiguration() + with open(path, 'r', encoding='utf-8') as f: + try: + data = json.load(f) + return MCPConfiguration.from_dict(data) + except Exception: + return MCPConfiguration() + diff --git a/tests/test_mcp_config.py b/tests/test_mcp_config.py new file mode 100644 index 0000000..0574321 --- /dev/null +++ b/tests/test_mcp_config.py @@ -0,0 +1,53 @@ +import os +import json +import pytest +from src import models + +def test_mcp_server_config_to_from_dict(): + data = { + "command": "node", + "args": ["server.js"], + "auto_start": True + } + cfg = models.MCPServerConfig.from_dict("test-server", data) + assert cfg.name == "test-server" + assert cfg.command == "node" + assert cfg.args == ["server.js"] + assert cfg.auto_start is True + + assert cfg.to_dict() == data + +def test_mcp_configuration_to_from_dict(): + data = { + "mcpServers": { + "server1": { + "command": "python", + "args": ["-m", "mcp_server"], + "auto_start": False + }, + "server2": { + "url": "http://localhost:8080/sse", + "auto_start": True + } + } + } + cfg = models.MCPConfiguration.from_dict(data) + assert len(cfg.mcpServers) == 2 + assert cfg.mcpServers["server1"].command == "python" + assert cfg.mcpServers["server2"].url == "http://localhost:8080/sse" + assert cfg.to_dict() == data + +def test_load_mcp_config(tmp_path): + config_file = tmp_path / "mcp_config.json" + data = { + "mcpServers": { + "test": {"command": "echo", "args": ["hello"]} + } + } + config_file.write_text(json.dumps(data)) + + # We'll need a way to load from a specific path + # Maybe models.load_mcp_config(path) + cfg = models.load_mcp_config(str(config_file)) + assert "test" in cfg.mcpServers + assert cfg.mcpServers["test"].command == "echo"