Private
Public Access
refactor(mcp_client): merge MCP config classes + load_mcp_config from models.py
Per the 4-criteria decision rule: MCP config classes (MCPServerConfig,
MCPConfiguration, VectorStoreConfig, RAGConfig) + load_mcp_config are
used by mcp_client + api_hooks + app_controller (3 systems) but
they are tightly coupled to the MCP subsystem's data layer. The test
file tests/test_mcp_config.py exists. Per the v2 spec: MERGE into
the existing src/mcp_client.py (the destination file IS the MCP
subsystem; the data layer belongs with the dispatcher).
This commit:
1. Adds MCPServerConfig + MCPConfiguration + VectorStoreConfig +
RAGConfig + load_mcp_config class/function definitions to
src/mcp_client.py at the top (after the imports + before the
mutating tools sentinel).
2. Removes the same class defs from src/models.py.
3. Adds lazy re-export via the existing __getattr__ in src/models.py
(EAGER would cycle: mcp_client was previously accessing them
via 'models.X'; eager re-export would deadlock).
4. Updates src/mcp_client.py internal references:
- 'def __init__(self, config: models.MCPServerConfig)' -> 'MCPServerConfig'
- 'async def add_server(self, config: models.MCPServerConfig)' -> 'MCPServerConfig'
Verification: VC8 (MCP config classes + load_mcp_config)
from src.mcp_client import MCPServerConfig, MCPConfiguration,
VectorStoreConfig, RAGConfig,
load_mcp_config # OK
from src.models import MCPServerConfig, MCPConfiguration,
VectorStoreConfig, RAGConfig,
load_mcp_config # OK (lazy)
identity check: True for all 5
Tests verified (4/4 PASS):
tests/test_mcp_config.py (3 tests)
tests/test_mcp_client_beads.py (1 test)
Consumer check (lazy __getattr__ keeps these working):
src/app_controller.py: models.MCPConfiguration, models.RAGConfig,
models.load_mcp_config (7+ sites)
src/rag_engine.py: models.RAGConfig (1 site)
All resolve via the lazy __getattr__.
This commit is contained in:
+123
-6
@@ -62,9 +62,10 @@ import subprocess
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
from typing import Optional, Callable, Any, cast
|
||||
from dataclasses import dataclass, field
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Callable, Any, cast
|
||||
|
||||
from scripts import py_struct_tools
|
||||
|
||||
@@ -73,7 +74,123 @@ from src import models
|
||||
from src import outline_tool
|
||||
from src import summarize
|
||||
from src import mcp_tool_specs
|
||||
from src.result_types import ErrorInfo, ErrorKind, NilPath, Result
|
||||
from src.result_types import ErrorInfo, ErrorKind, NilPath, Result
|
||||
from src.type_aliases import Metadata
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- MCP config dataclasses
|
||||
# Moved from src/models.py in module_taxonomy_refactor_20260627 Phase 3i.
|
||||
# These are the data layer of the MCP subsystem; they belong here.
|
||||
|
||||
@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) -> Metadata:
|
||||
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: Metadata) -> "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) -> Metadata:
|
||||
return {'mcpServers': {name: cfg.to_dict() for name, cfg in self.mcpServers.items()}}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Metadata) -> "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)
|
||||
|
||||
|
||||
@dataclass
|
||||
class VectorStoreConfig:
|
||||
provider: str
|
||||
url: Optional[str] = None
|
||||
api_key: Optional[str] = None
|
||||
collection_name: str = 'manual_slop'
|
||||
mcp_server: Optional[str] = None
|
||||
mcp_tool: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> Metadata:
|
||||
return {
|
||||
"provider": self.provider,
|
||||
"url": self.url,
|
||||
"api_key": self.api_key,
|
||||
"collection_name": self.collection_name,
|
||||
"mcp_server": self.mcp_server,
|
||||
"mcp_tool": self.mcp_tool,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Metadata) -> "VectorStoreConfig":
|
||||
return cls(
|
||||
provider = data["provider"],
|
||||
url = data.get("url"),
|
||||
api_key = data.get("api_key"),
|
||||
collection_name = data.get("collection_name", "manual_slop"),
|
||||
mcp_server = data.get("mcp_server"),
|
||||
mcp_tool = data.get("mcp_tool"),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RAGConfig:
|
||||
enabled: bool = False
|
||||
vector_store: VectorStoreConfig = field(default_factory=lambda: VectorStoreConfig(provider='mock'))
|
||||
embedding_provider: str = 'gemini'
|
||||
chunk_size: int = 1000
|
||||
chunk_overlap: int = 200
|
||||
|
||||
def to_dict(self) -> Metadata:
|
||||
return {
|
||||
"enabled": self.enabled,
|
||||
"vector_store": self.vector_store.to_dict(),
|
||||
"embedding_provider": self.embedding_provider,
|
||||
"chunk_size": self.chunk_size,
|
||||
"chunk_overlap": self.chunk_overlap,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Metadata) -> "RAGConfig":
|
||||
return cls(
|
||||
enabled = data.get("enabled", False),
|
||||
vector_store = VectorStoreConfig.from_dict(data.get("vector_store", {"provider": "mock"})),
|
||||
embedding_provider = data.get("embedding_provider", "gemini"),
|
||||
chunk_size = data.get("chunk_size", 1000),
|
||||
chunk_overlap = data.get("chunk_overlap", 200),
|
||||
)
|
||||
|
||||
|
||||
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 (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
|
||||
_mcp_err = Result(data=MCPConfiguration(), errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"failed to load MCP config: {e}", source="mcp_client.load_mcp_config", original=e)])
|
||||
return _mcp_err.data
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ mutating tools sentinel
|
||||
@@ -1621,7 +1738,7 @@ def get_ui_performance() -> str:
|
||||
# ------------------------------------------------------------------ tool dispatch
|
||||
|
||||
class StdioMCPServer:
|
||||
def __init__(self, config: models.MCPServerConfig):
|
||||
def __init__(self, config: MCPServerConfig):
|
||||
self.config = config
|
||||
self.name = config.name
|
||||
self.proc = None
|
||||
@@ -1720,7 +1837,7 @@ class ExternalMCPManager:
|
||||
"""Initialize the manager with an empty server registry."""
|
||||
self.servers = {}
|
||||
|
||||
async def add_server(self, config: models.MCPServerConfig):
|
||||
async def add_server(self, config: MCPServerConfig):
|
||||
"""
|
||||
Add and start a new MCP server from a configuration object.
|
||||
[C: tests/test_external_mcp.py:test_external_mcp_real_process, tests/test_external_mcp.py:test_get_tool_schemas_includes_external]
|
||||
|
||||
Reference in New Issue
Block a user