feat (mcp-conf): Project-level MCP configuration with merge/override strategy

src/kimi_cli/acp/server.py

@@ -8,6 +8,7 @@
 from typing import Any, NamedTuple
 
 import acp
+from fastmcp.mcp_config import MCPConfig
 from kaos.path import KaosPath
 
 from kimi_cli.acp.kaos import ACPKaos
@@ -18,6 +19,7 @@
 from kimi_cli.acp.version import ACPVersionSpec, negotiate_version
 from kimi_cli.app import KimiCLI
 from kimi_cli.auth.oauth import KIMI_CODE_OAUTH_KEY, load_tokens
+from kimi_cli.cli.mcp import collect_file_mcp_configs
 from kimi_cli.config import LLMModel, OAuthRef, load_config, save_config
 from kimi_cli.constant import NAME, VERSION
 from kimi_cli.llm import create_llm, derive_model_capabilities
@@ -27,6 +29,41 @@
 from kimi_cli.utils.logging import logger
 
 
+def _collect_acp_mcp_configs(
+    cwd: str, mcp_servers: list[MCPServer] | None = None
+) -> list[MCPConfig | dict[str, Any]]:
+    """Collect MCP configs for ACP session, respecting merge_strategy."""
+    from kimi_cli.config import load_config
+
+    strategy = load_config().mcp.merge_strategy
+    configs: list[MCPConfig | dict[str, Any]] = []
+
+    if strategy == "merge":
+        try:
+            file_configs = collect_file_mcp_configs(strategy, work_dir=Path(cwd))
+            configs.extend(file_configs)
+        except Exception as exc:
+            logger.warning(
+                "Failed to load file-based MCP configs for ACP session: {error}",
+                error=exc,
+            )
+        configs.append(acp_mcp_servers_to_mcp_config(mcp_servers or []))
+    else:  # override
+        if mcp_servers:
+            configs.append(acp_mcp_servers_to_mcp_config(mcp_servers))
+        else:
+            try:
+                file_configs = collect_file_mcp_configs(strategy, work_dir=Path(cwd))
+                configs.extend(file_configs)
+            except Exception as exc:
+                logger.warning(
+                    "Failed to load file-based MCP configs for ACP session: {error}",
+                    error=exc,
+                )
+
+    return configs
+
+
 class ACPServer:
     def __init__(self) -> None:
         self.client_capabilities: acp.schema.ClientCapabilities | None = None
@@ -159,10 +196,10 @@ async def new_session(
 
         session = await Session.create(KaosPath.unsafe_from_local_path(Path(cwd)))
 
-        mcp_config = acp_mcp_servers_to_mcp_config(mcp_servers or [])
+        mcp_configs = _collect_acp_mcp_configs(cwd, mcp_servers)
         cli_instance = await KimiCLI.create(
             session,
-            mcp_configs=[mcp_config],
+            mcp_configs=mcp_configs,
             ui_mode="acp",
         )
         config = cli_instance.soul.runtime.config
@@ -229,10 +266,10 @@ async def _setup_session(
             )
             raise acp.RequestError.invalid_params({"session_id": "Session not found"})
 
-        mcp_config = acp_mcp_servers_to_mcp_config(mcp_servers or [])
+        mcp_configs = _collect_acp_mcp_configs(cwd, mcp_servers)
         cli_instance = await KimiCLI.create(
             session,
-            mcp_configs=[mcp_config],
+            mcp_configs=mcp_configs,
             resumed=True,  # _setup_session loads existing sessions
             ui_mode="acp",
         )

src/kimi_cli/app.py

@@ -6,7 +6,7 @@
 import sys
 import time
 import warnings
-from collections.abc import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Callable, Sequence
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -134,7 +134,7 @@ async def create(
         ui_mode: str = "shell",
         # Extensions
         agent_file: Path | None = None,
-        mcp_configs: list[MCPConfig] | list[dict[str, Any]] | None = None,
+        mcp_configs: Sequence[MCPConfig | dict[str, Any]] | None = None,
         skills_dirs: list[KaosPath] | None = None,
         # Loop control
         max_steps_per_turn: int | None = None,

src/kimi_cli/cli/__init__.py

@@ -309,7 +309,8 @@ def kimi(
             readable=True,
             help=(
                 "MCP config file to load. Add this option multiple times to specify multiple MCP "
-                "configs. Default: none."
+                "configs. When omitted, the project-level `<work_dir>/.kimi/mcp.json` is used if "
+                "it exists; otherwise the global `~/.kimi/mcp.json` is used as a fallback."
             ),
         ),
     ] = None,
@@ -366,7 +367,6 @@ def kimi(
     """Kimi, your next CLI agent."""
     import asyncio
     import contextlib
-    import json
 
     from kimi_cli.utils.proctitle import init_process_name
 
@@ -381,15 +381,15 @@ def kimi(
 
     from kimi_cli.agentspec import DEFAULT_AGENT_FILE, OKABE_AGENT_FILE
     from kimi_cli.app import KimiCLI, enable_logging
-    from kimi_cli.config import Config, load_config_from_string
-    from kimi_cli.exception import ConfigError
+    from kimi_cli.config import Config, load_config, load_config_from_string
+    from kimi_cli.exception import ConfigError, MCPConfigError
     from kimi_cli.hooks import events as hook_events
     from kimi_cli.metadata import load_metadata, save_metadata
     from kimi_cli.session import Session
     from kimi_cli.ui.shell.startup import ShellStartupProgress
     from kimi_cli.utils.logging import logger, open_original_stderr, redirect_stderr_to_logger
 
-    from .mcp import get_global_mcp_config_file
+    from .mcp import collect_file_mcp_configs
 
     # Don't redirect stderr during argument parsing. Our stderr redirector
     # replaces fd=2 with a pipe, which would swallow Click/Typer startup errors.
@@ -515,28 +515,31 @@ def _emit_fatal_error(message: str) -> None:
     file_configs = list(mcp_config_file or [])
     raw_mcp_config = list(mcp_config or [])
 
-    # Use default MCP config file if no MCP config is provided
-    if not file_configs:
-        default_mcp_file = get_global_mcp_config_file()
-        if default_mcp_file.exists():
-            file_configs.append(default_mcp_file)
+    work_dir = KaosPath.unsafe_from_local_path(local_work_dir) if local_work_dir else KaosPath.cwd()
 
-    try:
-        mcp_configs = [json.loads(conf.read_text(encoding="utf-8")) for conf in file_configs]
-    except json.JSONDecodeError as e:
-        raise typer.BadParameter(f"Invalid JSON: {e}", param_hint="--mcp-config-file") from e
+    # Resolve MCP merge strategy from config (load default config if needed).
+    if isinstance(config, Config):
+        mcp_strategy = config.mcp.merge_strategy
+    elif config is not None:
+        mcp_strategy = load_config(config).mcp.merge_strategy
+    else:
+        config = load_config()
+        mcp_strategy = config.mcp.merge_strategy
 
     try:
-        mcp_configs += [json.loads(conf) for conf in raw_mcp_config]
-    except json.JSONDecodeError as e:
-        raise typer.BadParameter(f"Invalid JSON: {e}", param_hint="--mcp-config") from e
+        mcp_configs = collect_file_mcp_configs(
+            mcp_strategy,
+            work_dir=work_dir.unsafe_to_local_path(),
+            explicit_files=file_configs,
+            raw_jsons=raw_mcp_config,
+        )
+    except MCPConfigError as e:
+        raise typer.BadParameter(str(e), param_hint="--mcp-config-file") from e
 
     skills_dirs: list[KaosPath] | None = None
     if local_skills_dir:
         skills_dirs = [KaosPath.unsafe_from_local_path(p) for p in local_skills_dir]
 
-    work_dir = KaosPath.unsafe_from_local_path(local_work_dir) if local_work_dir else KaosPath.cwd()
-
     # Tracks the most recently created/loaded session so that _reload_loop's
     # exception handler can clean it up even when _run() fails before returning.
     _latest_created_session: Session | None = None

src/kimi_cli/cli/mcp.py

@@ -4,6 +4,8 @@
 
 import typer
 
+from kimi_cli.exception import MCPConfigError
+
 cli = typer.Typer(help="Manage MCP server configurations.")
 
 
@@ -14,6 +16,106 @@ def get_global_mcp_config_file() -> Path:
     return get_share_dir() / "mcp.json"
 
 
+def get_project_mcp_config_file(work_dir: Path) -> Path | None:
+    """Get the project-level MCP config file path if it exists.
+
+    Args:
+        work_dir: The working directory to search for `.kimi/mcp.json`.
+
+    Returns:
+        Path to the project-level MCP config file if it exists, otherwise None.
+    """
+    project_mcp_file = work_dir / ".kimi" / "mcp.json"
+    if project_mcp_file.exists():
+        return project_mcp_file
+    return None
+
+
+def collect_file_mcp_configs(
+    strategy: Literal["merge", "override"],
+    *,
+    work_dir: Path | None = None,
+    explicit_files: list[Path] | None = None,
+    raw_jsons: list[str] | None = None,
+) -> list[dict[str, Any]]:
+    """Collect MCP config dicts from file sources according to the strategy.
+
+    Priority (lowest to highest):
+        1. Global ``~/.kimi/mcp.json``
+        2. Project ``<work_dir>/.kimi/mcp.json``
+        3. Explicit ``--mcp-config-file`` paths
+        4. Raw ``--mcp-config`` JSON strings
+
+    For ``"merge"`` all available layers are returned in priority order.
+    Later layers override duplicate server names when the configs are merged.
+
+    For ``"override"`` only the highest-priority available layer is returned.
+    Explicit inputs beat project config, which beats global config.
+
+    Args:
+        strategy: ``"merge"`` or ``"override"``.
+        work_dir: Working directory to search for project-level config.
+        explicit_files: Paths explicitly provided by the user.
+        raw_jsons: Raw JSON strings explicitly provided by the user.
+
+    Returns:
+        List of parsed MCP config dicts.
+    """
+    explicit_files = explicit_files or []
+    raw_jsons = raw_jsons or []
+
+    def _load(path: Path) -> dict[str, Any]:
+        try:
+            return json.loads(path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError as exc:
+            raise MCPConfigError(f"Invalid JSON in MCP config file '{path}': {exc}") from exc
+
+    configs: list[dict[str, Any]] = []
+
+    if strategy == "merge":
+        global_file = get_global_mcp_config_file()
+        if global_file.exists():
+            configs.append(_load(global_file))
+
+        if work_dir is not None:
+            project_file = get_project_mcp_config_file(work_dir)
+            if project_file is not None:
+                configs.append(_load(project_file))
+
+        for path in explicit_files:
+            configs.append(_load(path))
+
+        for text in raw_jsons:
+            try:
+                configs.append(json.loads(text))
+            except json.JSONDecodeError as exc:
+                raise MCPConfigError(f"Invalid JSON in --mcp-config: {exc}") from exc
+
+    else:  # override
+        if explicit_files or raw_jsons:
+            for path in explicit_files:
+                configs.append(_load(path))
+            for text in raw_jsons:
+                try:
+                    configs.append(json.loads(text))
+                except json.JSONDecodeError as exc:
+                    raise MCPConfigError(f"Invalid JSON in --mcp-config: {exc}") from exc
+        elif work_dir is not None:
+            project_file = get_project_mcp_config_file(work_dir)
+            if project_file is not None:
+                configs.append(_load(project_file))
+            else:
+                global_file = get_global_mcp_config_file()
+                if global_file.exists():
+                    configs.append(_load(global_file))
+        else:
+            global_file = get_global_mcp_config_file()
+            if global_file.exists():
+                configs.append(_load(global_file))
+
+    return configs
+
+
 def _load_mcp_config() -> dict[str, Any]:
     """Load MCP config from global mcp config file."""
     from fastmcp.mcp_config import MCPConfig

src/kimi_cli/config.py

@@ -181,6 +181,16 @@ class MCPConfig(BaseModel):
     client: MCPClientConfig = Field(
         default_factory=MCPClientConfig, description="MCP client configuration"
     )
+    merge_strategy: Literal["merge", "override"] = Field(
+        default="override",
+        description=(
+            "How to combine MCP configs from multiple sources. "
+            "'merge' combines global, project-level and explicit configs, "
+            "with later sources overriding duplicate server names. "
+            "'override' uses only the highest-priority source that is defined. "
+            "Default is 'override'."
+        ),
+    )
 
 
 class Config(BaseModel):

src/kimi_cli/soul/agent.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Sequence
 from dataclasses import asdict, dataclass
 from datetime import datetime
 from pathlib import Path
@@ -384,7 +385,7 @@ async def load_agent(
     agent_file: Path,
     runtime: Runtime,
     *,
-    mcp_configs: list[MCPConfig] | list[dict[str, Any]],
+    mcp_configs: Sequence[MCPConfig | dict[str, Any]],
     start_mcp_loading: bool = True,
 ) -> Agent:
     """

src/kimi_cli/web/runner/worker.py

@@ -10,14 +10,12 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import sys
-from typing import Any
 from uuid import UUID
 
 from kimi_cli import logger
 from kimi_cli.app import KimiCLI, enable_logging
-from kimi_cli.cli.mcp import get_global_mcp_config_file
+from kimi_cli.cli.mcp import collect_file_mcp_configs
 from kimi_cli.exception import MCPConfigError
 from kimi_cli.web.store.sessions import load_session_by_id
 
@@ -32,18 +30,20 @@ async def run_worker(session_id: UUID) -> None:
     # Get the kimi-cli session object
     session = joint_session.kimi_cli_session
 
-    # Load default MCP config file if it exists
-    default_mcp_file = get_global_mcp_config_file()
-    mcp_configs: list[dict[str, Any]] = []
-    if default_mcp_file.exists():
-        raw = default_mcp_file.read_text(encoding="utf-8")
-        try:
-            mcp_configs = [json.loads(raw)]
-        except json.JSONDecodeError:
-            logger.warning(
-                "Invalid JSON in MCP config file: {path}",
-                path=default_mcp_file,
-            )
+    # Load MCP config files according to merge_strategy.
+    work_dir = session.dir
+
+    from kimi_cli.config import load_config
+
+    strategy = load_config().mcp.merge_strategy
+    try:
+        mcp_configs = collect_file_mcp_configs(strategy, work_dir=work_dir)
+    except Exception as exc:
+        logger.warning(
+            "Failed to load MCP configs for web worker: {error}",
+            error=exc,
+        )
+        mcp_configs = []
 
     # Detect whether this is a resumed session (has prior state on disk)
     # vs a brand-new session that should honor config.default_plan_mode.
@@ -56,8 +56,7 @@ async def run_worker(session_id: UUID) -> None:
         )
     except MCPConfigError as exc:
         logger.warning(
-            "Invalid MCP config in {path}: {error}. Starting without MCP.",
-            path=default_mcp_file,
+            "Invalid MCP config: {error}. Starting without MCP.",
             error=exc,
         )
         kimi_cli = await KimiCLI.create(session, mcp_configs=None, resumed=resumed, ui_mode="wire")