From 6439227e42d161772153cf7f4fe5ed62faa948f7 Mon Sep 17 00:00:00 2001 From: Alexandr Anisimov Date: Sat, 23 May 2026 09:41:36 +0300 Subject: [PATCH 1/2] Add the ability to override mcp configurations at the project level with the merge/overwrute strategy. --- src/kimi_cli/acp/server.py | 45 +++++++++++-- src/kimi_cli/app.py | 4 +- src/kimi_cli/cli/__init__.py | 41 ++++++------ src/kimi_cli/cli/mcp.py | 102 ++++++++++++++++++++++++++++++ src/kimi_cli/config.py | 9 +++ src/kimi_cli/soul/agent.py | 3 +- src/kimi_cli/web/runner/worker.py | 33 +++++----- tests/core/test_plan_flag.py | 4 +- 8 files changed, 196 insertions(+), 45 deletions(-) diff --git a/src/kimi_cli/acp/server.py b/src/kimi_cli/acp/server.py index 0b6b32538..aa1432825 100644 --- a/src/kimi_cli/acp/server.py +++ b/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", ) diff --git a/src/kimi_cli/app.py b/src/kimi_cli/app.py index fb3b6ab31..5c06c8c31 100644 --- a/src/kimi_cli/app.py +++ b/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, diff --git a/src/kimi_cli/cli/__init__.py b/src/kimi_cli/cli/__init__.py index b9aafb7a8..fc654d039 100644 --- a/src/kimi_cli/cli/__init__.py +++ b/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 `/.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 diff --git a/src/kimi_cli/cli/mcp.py b/src/kimi_cli/cli/mcp.py index 61ad2b1fd..3e0e42674 100644 --- a/src/kimi_cli/cli/mcp.py +++ b/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 ``/.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 diff --git a/src/kimi_cli/config.py b/src/kimi_cli/config.py index 451f46810..90cd22f8d 100644 --- a/src/kimi_cli/config.py +++ b/src/kimi_cli/config.py @@ -181,6 +181,15 @@ class MCPConfig(BaseModel): client: MCPClientConfig = Field( default_factory=MCPClientConfig, description="MCP client configuration" ) + merge_strategy: Literal["merge", "override"] = Field( + default="merge", + 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." + ), + ) class Config(BaseModel): diff --git a/src/kimi_cli/soul/agent.py b/src/kimi_cli/soul/agent.py index a208e0879..d165e5413 100644 --- a/src/kimi_cli/soul/agent.py +++ b/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: """ diff --git a/src/kimi_cli/web/runner/worker.py b/src/kimi_cli/web/runner/worker.py index 807095bc9..65994cd5a 100644 --- a/src/kimi_cli/web/runner/worker.py +++ b/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") diff --git a/tests/core/test_plan_flag.py b/tests/core/test_plan_flag.py index 22adaa79c..24b7f8506 100644 --- a/tests/core/test_plan_flag.py +++ b/tests/core/test_plan_flag.py @@ -329,7 +329,7 @@ async def spy_create(session, **kwargs): with ( patch("kimi_cli.web.runner.worker.load_session_by_id", return_value=fake_joint), patch( - "kimi_cli.web.runner.worker.get_global_mcp_config_file", + "kimi_cli.cli.mcp.get_global_mcp_config_file", return_value=tmp_path / "no-mcp.json", ), patch.object(KimiCLI, "create", side_effect=spy_create), @@ -363,7 +363,7 @@ async def spy_create(session, **kwargs): with ( patch("kimi_cli.web.runner.worker.load_session_by_id", return_value=fake_joint), patch( - "kimi_cli.web.runner.worker.get_global_mcp_config_file", + "kimi_cli.cli.mcp.get_global_mcp_config_file", return_value=tmp_path / "no-mcp.json", ), patch.object(KimiCLI, "create", side_effect=spy_create), From fe2ba8e679ca11d90f44b0fe626fd538f980ec08 Mon Sep 17 00:00:00 2001 From: Alexandr Anisimov Date: Sat, 23 May 2026 10:13:27 +0300 Subject: [PATCH 2/2] fix(config): change default MCP merge_strategy from merge to override --- src/kimi_cli/config.py | 5 +- tests/cli/test_project_mcp_config.py | 421 +++++++++++++++++++++++++++ 2 files changed, 424 insertions(+), 2 deletions(-) create mode 100644 tests/cli/test_project_mcp_config.py diff --git a/src/kimi_cli/config.py b/src/kimi_cli/config.py index 90cd22f8d..87d5bf725 100644 --- a/src/kimi_cli/config.py +++ b/src/kimi_cli/config.py @@ -182,12 +182,13 @@ class MCPConfig(BaseModel): default_factory=MCPClientConfig, description="MCP client configuration" ) merge_strategy: Literal["merge", "override"] = Field( - default="merge", + 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." + "'override' uses only the highest-priority source that is defined. " + "Default is 'override'." ), ) diff --git a/tests/cli/test_project_mcp_config.py b/tests/cli/test_project_mcp_config.py new file mode 100644 index 000000000..56b1a7ba8 --- /dev/null +++ b/tests/cli/test_project_mcp_config.py @@ -0,0 +1,421 @@ +"""Tests for project-level MCP config loading (.kimi/mcp.json).""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from typer.testing import CliRunner + +from kimi_cli.cli import cli +from kimi_cli.cli.mcp import collect_file_mcp_configs, get_project_mcp_config_file + +# --------------------------------------------------------------------------- +# get_project_mcp_config_file +# --------------------------------------------------------------------------- + + +def test_get_project_mcp_config_file_returns_path_when_exists(tmp_path: Path) -> None: + mcp_file = tmp_path / ".kimi" / "mcp.json" + mcp_file.parent.mkdir(parents=True) + mcp_file.write_text("{}", encoding="utf-8") + + result = get_project_mcp_config_file(tmp_path) + + assert result == mcp_file + + +def test_get_project_mcp_config_file_returns_none_when_missing(tmp_path: Path) -> None: + result = get_project_mcp_config_file(tmp_path) + assert result is None + + +# --------------------------------------------------------------------------- +# collect_file_mcp_configs +# --------------------------------------------------------------------------- + + +def test_collect_merge_includes_global_and_project(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + project_mcp = work_dir / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"p": {"command": "p"}}})) + + configs = collect_file_mcp_configs("merge", work_dir=work_dir) + + assert len(configs) == 2 + assert configs[0]["mcpServers"]["g"]["command"] == "g" + assert configs[1]["mcpServers"]["p"]["command"] == "p" + + +def test_collect_merge_project_overrides_global_duplicate(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"srv": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + project_mcp = work_dir / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"srv": {"command": "p"}}})) + + configs = collect_file_mcp_configs("merge", work_dir=work_dir) + + # In merge mode both configs are returned; the toolset will overwrite duplicates + assert len(configs) == 2 + assert configs[0]["mcpServers"]["srv"]["command"] == "g" + assert configs[1]["mcpServers"]["srv"]["command"] == "p" + + +def test_collect_override_uses_only_project_when_present(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + project_mcp = work_dir / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"p": {"command": "p"}}})) + + configs = collect_file_mcp_configs("override", work_dir=work_dir) + + assert len(configs) == 1 + assert configs[0]["mcpServers"]["p"]["command"] == "p" + + +def test_collect_override_fallback_to_global_when_no_project(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + + configs = collect_file_mcp_configs("override", work_dir=work_dir) + + assert len(configs) == 1 + assert configs[0]["mcpServers"]["g"]["command"] == "g" + + +def test_collect_override_prefers_explicit_files(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + project_mcp = work_dir / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"p": {"command": "p"}}})) + + explicit = tmp_path / "exp.json" + explicit.write_text(json.dumps({"mcpServers": {"e": {"command": "e"}}})) + + configs = collect_file_mcp_configs("override", work_dir=work_dir, explicit_files=[explicit]) + + assert len(configs) == 1 + assert configs[0]["mcpServers"]["e"]["command"] == "e" + + +def test_collect_merge_includes_explicit_and_raw_on_top(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + work_dir = tmp_path / "work" + work_dir.mkdir() + + explicit = tmp_path / "exp.json" + explicit.write_text(json.dumps({"mcpServers": {"e": {"command": "e"}}})) + + configs = collect_file_mcp_configs( + "merge", + work_dir=work_dir, + explicit_files=[explicit], + raw_jsons=[json.dumps({"mcpServers": {"r": {"command": "r"}}})], + ) + + assert len(configs) == 3 + names = [list(c["mcpServers"].keys())[0] for c in configs] + assert names == ["g", "e", "r"] + + +# --------------------------------------------------------------------------- +# CLI integration tests +# --------------------------------------------------------------------------- + + +def _patch_kimi_cli_create(monkeypatch): + """Patch KimiCLI.create to capture mcp_configs without doing real I/O.""" + calls: list[dict] = [] + + async def fake_create(session, *, mcp_configs=None, **kwargs): + calls.append({"mcp_configs": mcp_configs, **kwargs}) + return SimpleNamespace( + soul=SimpleNamespace( + runtime=SimpleNamespace(config=SimpleNamespace(default_model="test")), + hook_engine=SimpleNamespace(trigger=AsyncMock()), + ), + run_print=AsyncMock(return_value=0), + shutdown_background_tasks=AsyncMock(), + await_bg_tasks_shutdown=AsyncMock(), + ) + + monkeypatch.setattr("kimi_cli.app.KimiCLI.create", fake_create) + return calls + + +def test_cli_uses_project_mcp_config_when_no_explicit_file(tmp_path: Path, monkeypatch) -> None: + # Isolate share dir so the real ~/.kimi/mcp.json is not picked up. + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + calls = _patch_kimi_cli_create(monkeypatch) + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"proj": {"command": "proj"}}})) + + result = CliRunner().invoke( + cli, + ["--work-dir", str(tmp_path), "--print", "--prompt", "hello"], + ) + + assert result.exit_code == 0, result.output + assert len(calls) == 1 + mcp_configs = calls[0]["mcp_configs"] + assert len(mcp_configs) == 1 + assert mcp_configs[0]["mcpServers"]["proj"]["command"] == "proj" + + +def test_cli_prefers_explicit_mcp_config_file_over_project(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("KIMI_SHARE_DIR", str(tmp_path / "share")) + calls = _patch_kimi_cli_create(monkeypatch) + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"proj": {"command": "proj"}}})) + + explicit_mcp = tmp_path / "explicit.json" + explicit_mcp.write_text(json.dumps({"mcpServers": {"exp": {"command": "exp"}}})) + + result = CliRunner().invoke( + cli, + [ + "--work-dir", + str(tmp_path), + "--mcp-config-file", + str(explicit_mcp), + "--print", + "--prompt", + "hello", + ], + ) + + assert result.exit_code == 0, result.output + assert len(calls) == 1 + mcp_configs = calls[0]["mcp_configs"] + # In merge mode (default) explicit files are appended on top of project + assert len(mcp_configs) == 2 + assert mcp_configs[0]["mcpServers"]["proj"]["command"] == "proj" + assert mcp_configs[1]["mcpServers"]["exp"]["command"] == "exp" + + +def test_cli_falls_back_to_global_mcp_config_when_no_project_config( + tmp_path: Path, monkeypatch +) -> None: + calls = _patch_kimi_cli_create(monkeypatch) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"global": {"command": "global"}}})) + + monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir)) + + work_dir = tmp_path / "work" + work_dir.mkdir() + + result = CliRunner().invoke( + cli, + ["--work-dir", str(work_dir), "--print", "--prompt", "hello"], + ) + + assert result.exit_code == 0, result.output + assert len(calls) == 1 + mcp_configs = calls[0]["mcp_configs"] + assert len(mcp_configs) == 1 + assert mcp_configs[0]["mcpServers"]["global"]["command"] == "global" + + +def test_cli_ignores_invalid_project_mcp_config(tmp_path: Path, monkeypatch) -> None: + _patch_kimi_cli_create(monkeypatch) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"global": {"command": "global"}}})) + + monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir)) + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text("not json") + + result = CliRunner().invoke( + cli, + ["--work-dir", str(tmp_path), "--print", "--prompt", "hello"], + ) + + # Invalid JSON should raise a BadParameter error before reaching create() + assert result.exit_code != 0 + assert "Invalid JSON" in result.output + + +def test_cli_merge_strategy_merges_global_and_project(tmp_path: Path, monkeypatch) -> None: + calls = _patch_kimi_cli_create(monkeypatch) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir)) + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"p": {"command": "p"}}})) + + config_file = tmp_path / "config.toml" + config_file.write_text('[mcp]\nmerge_strategy = "merge"\n') + + result = CliRunner().invoke( + cli, + [ + "--work-dir", + str(tmp_path), + "--config-file", + str(config_file), + "--print", + "--prompt", + "hello", + ], + ) + + assert result.exit_code == 0, result.output + assert len(calls) == 1 + mcp_configs = calls[0]["mcp_configs"] + assert len(mcp_configs) == 2 + assert mcp_configs[0]["mcpServers"]["g"]["command"] == "g" + assert mcp_configs[1]["mcpServers"]["p"]["command"] == "p" + + +def test_cli_override_strategy_uses_only_project(tmp_path: Path, monkeypatch) -> None: + calls = _patch_kimi_cli_create(monkeypatch) + share_dir = tmp_path / "share" + share_dir.mkdir() + global_mcp = share_dir / "mcp.json" + global_mcp.write_text(json.dumps({"mcpServers": {"g": {"command": "g"}}})) + + monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir)) + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"p": {"command": "p"}}})) + + config_file = tmp_path / "config.toml" + config_file.write_text('[mcp]\nmerge_strategy = "override"\n') + + result = CliRunner().invoke( + cli, + [ + "--work-dir", + str(tmp_path), + "--config-file", + str(config_file), + "--print", + "--prompt", + "hello", + ], + ) + + assert result.exit_code == 0, result.output + assert len(calls) == 1 + mcp_configs = calls[0]["mcp_configs"] + assert len(mcp_configs) == 1 + assert mcp_configs[0]["mcpServers"]["p"]["command"] == "p" + + +# --------------------------------------------------------------------------- +# ACP server integration tests +# --------------------------------------------------------------------------- + + +@pytest.fixture +def isolated_acp_share_dir(monkeypatch, tmp_path: Path): + """Isolate ACP tests from the real ~/.kimi directory.""" + share_dir = tmp_path / "share" + share_dir.mkdir() + monkeypatch.setenv("KIMI_SHARE_DIR", str(share_dir)) + return share_dir + + +@pytest.mark.asyncio +async def test_acp_server_loads_project_mcp_config( + tmp_path: Path, isolated_acp_share_dir: Path +) -> None: + from kimi_cli.acp.server import _collect_acp_mcp_configs + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text(json.dumps({"mcpServers": {"proj": {"command": "proj"}}})) + + configs = _collect_acp_mcp_configs(str(tmp_path)) + + # merge mode: project config + empty ACP config + assert len(configs) == 2 + assert isinstance(configs[0], dict) + assert configs[0]["mcpServers"]["proj"]["command"] == "proj" + + +@pytest.mark.asyncio +async def test_acp_server_skips_invalid_project_mcp_config( + tmp_path: Path, isolated_acp_share_dir: Path +) -> None: + from kimi_cli.acp.server import _collect_acp_mcp_configs + + project_mcp = tmp_path / ".kimi" / "mcp.json" + project_mcp.parent.mkdir(parents=True) + project_mcp.write_text("not json") + + configs = _collect_acp_mcp_configs(str(tmp_path)) + + # invalid project skipped, only empty ACP config remains + assert len(configs) == 1 + + +@pytest.mark.asyncio +async def test_acp_server_returns_empty_when_no_project_config( + tmp_path: Path, isolated_acp_share_dir: Path +) -> None: + from kimi_cli.acp.server import _collect_acp_mcp_configs + + configs = _collect_acp_mcp_configs(str(tmp_path)) + # no file configs + empty ACP config + assert len(configs) == 1