feat(models): Add MCP configuration models and loading logic
This commit is contained in:
@@ -37,6 +37,8 @@ See Also:
|
|||||||
- src/project_manager.py for persistence layer
|
- src/project_manager.py for persistence layer
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
import json
|
||||||
|
import os
|
||||||
import tomllib
|
import tomllib
|
||||||
import datetime
|
import datetime
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
@@ -515,3 +517,56 @@ class Persona:
|
|||||||
bias_profile=data.get("bias_profile"),
|
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()
|
||||||
|
|
||||||
|
|||||||
53
tests/test_mcp_config.py
Normal file
53
tests/test_mcp_config.py
Normal file
@@ -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"
|
||||||
Reference in New Issue
Block a user