config: [mcp] section with server definitions and disable gate

This commit is contained in:
2026-09-08 11:53:19 +08:00
parent 7c377a1a53
commit dd70d02491
5 changed files with 153 additions and 0 deletions
+2
View File
@@ -13,6 +13,8 @@ from .models import DatabaseConfig as DatabaseConfig
from .models import DeepSeekConvention as DeepSeekConvention
from .models import DeepseekProvider as DeepseekProvider
from .models import HttpTimeoutConfig as HttpTimeoutConfig
from .models import McpConfig as McpConfig
from .models import McpServerConfig as McpServerConfig
from .models import ModelConfig as ModelConfig
from .models import NetworkingConfig as NetworkingConfig
from .models import OpenAICompatibleProvider as OpenAICompatibleProvider
+38
View File
@@ -143,6 +143,44 @@ class PluginsConfig(Struct, omit_defaults=True):
disable: list[str] = field(default_factory=list)
class McpServerConfig(Struct, omit_defaults=True):
"""One MCP server definition (stdio transport today).
TOML example::
[mcp.servers.docs]
command = "npx"
args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
env = { RUST_LOG = "info" }
# cwd = "/path" # spawn working directory (empty = inherit)
# timeout = 30.0 # per-request timeout in seconds
# read_only = true # mark every tool READ_ONLY (safe side turns)
``url`` is reserved for the streamable-HTTP transport; empty = stdio.
A defined server is connected unless its name appears in ``[mcp].disable``.
"""
command: str = ""
args: list[str] = field(default_factory=list)
env: dict[str, str] = field(default_factory=dict)
cwd: str = ""
timeout: float = 30.0
read_only: bool = False
url: str = ""
class McpConfig(Struct, omit_defaults=True):
"""Model Context Protocol servers exposed as agent tools.
Defining a server under ``[mcp.servers.<name>]`` is the opt-in; tools are
namespaced ``mcp__<server>__<tool>`` and registered with LOCAL tags. Names
in ``disable`` stay disconnected even when defined.
"""
servers: dict[str, McpServerConfig] = field(default_factory=dict)
disable: list[str] = field(default_factory=list)
class NetworkingConfig(Struct, omit_defaults=True):
"""Host-side network policy for tools such as ``fetch``.
+34
View File
@@ -9,6 +9,7 @@ import tomlkit
from .models import (
AgentConfig,
DatabaseConfig,
McpConfig,
ModelConfig,
NetworkingConfig,
PluginsConfig,
@@ -56,6 +57,29 @@ def _parse_networking(raw: dict[str, object]) -> NetworkingConfig:
return NetworkingConfig()
def _parse_mcp(raw: dict[str, object]) -> McpConfig:
"""Parse the ``[mcp]`` section, falling back to defaults.
Unknown keys (typos like ``cmds``) are rejected the same way as providers:
a malformed section silently degrades to "no MCP servers" instead of a
half-parsed one.
"""
try:
config = msgspec.convert(raw, McpConfig)
except msgspec.ValidationError:
return McpConfig()
known = {"servers", "disable"}
if set(raw) - known:
return McpConfig()
known_server = {"command", "args", "env", "cwd", "timeout", "read_only", "url"}
servers_raw = cast("dict[str, object]", raw.get("servers", {}))
for name, raw_entry in servers_raw.items():
entry = cast("dict[str, object]", raw_entry)
if not isinstance(raw_entry, dict) or not name.strip() or set(entry) - known_server:
return McpConfig()
return config
def _parse_providers(
document: tomlkit.TOMLDocument,
) -> tuple[dict[str, Provider], dict[str, object], dict[str, Provider]]:
@@ -125,6 +149,7 @@ class ConfigStore:
_agent: AgentConfig
_plugins: PluginsConfig
_networking: NetworkingConfig
_mcp: McpConfig
_providers: dict[str, Provider]
_bad_providers: dict[str, object]
_recoverable_providers: dict[str, Provider]
@@ -137,6 +162,7 @@ class ConfigStore:
self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(document)
@property
@@ -187,6 +213,13 @@ class ConfigStore:
"""Typed networking section (e.g. Fake-IP SSRF exemptions)."""
return self._networking
# -- mcp (read-only) --
@property
def mcp_config(self) -> McpConfig:
"""Typed MCP section (server definitions + disable list)."""
return self._mcp
def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig:
"""Replace the plugin enable list (in-memory; call :meth:`write` to persist)."""
cleaned = [name.strip() for name in names if name.strip()]
@@ -390,6 +423,7 @@ class ConfigStore:
self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(self._document)
# -- internal sync helpers --