Kimi-CLI 插件/扩展系统架构

技术研究人工智能 LLM

Kimi-CLI 提供了三层扩展机制： 1. 工具 - 基础功能单元 2. 技能 - 可复用的能力组合 3. MCP 服务器 - 外部工具协议

概述

Kimi-CLI 提供了三层扩展机制：

工具 - 基础功能单元
技能 - 可复用的能力组合
MCP 服务器 - 外部工具协议

1. 工具系统

1.1 工具定义

工具是 Agent 可调用的最小功能单元。工具继承自 CallableTool 或 CallableTool2 基类：

from kosong.tooling import CallableTool2, ToolReturnValue
from pydantic import BaseModel, Field

class Params(BaseModel):
    query: str = Field(description="搜索查询文本")
    limit: int = Field(default=5, ge=1, le=20)

class SearchWeb(CallableTool2[Params]):
    name: str = "SearchWeb"
    description: str = "搜索互联网获取最新信息"
    params: type[Params] = Params

    def __init__(self, config: Config, runtime: Runtime):
        super().__init__()
        self._runtime = runtime
        self._base_url = config.services.moonshot_search.base_url

    async def __call__(self, params: Params) -> ToolReturnValue:
        # 工具实现
        pass

1.2 工具加载机制

工具通过路径字符串加载，格式为 module.path:ClassName：

# 在 agent.yaml 中配置
tools:
  - "kimi_cli.tools.multiagent:Task"
  - "kimi_cli.tools.shell:Shell"
  - "kimi_cli.tools.file:ReadFile"

加载过程 (src/kimi_cli/soul/toolset.py):

def load_tools(self, tool_paths: list[str], dependencies: dict[type[Any], Any]) -> None:
    for tool_path in tool_paths:
        module_name, class_name = tool_path.rsplit(":", 1)
        module = importlib.import_module(module_name)
        tool_cls = getattr(module, class_name, None)
        
        # 依赖注入
        args: list[Any] = []
        if "__init__" in tool_cls.__dict__:
            for param in inspect.signature(tool_cls).parameters.values():
                if param.kind == inspect.Parameter.KEYWORD_ONLY:
                    break
                if param.annotation not in dependencies:
                    raise ValueError(f"Tool dependency not found: {param.annotation}")
                args.append(dependencies[param.annotation])
        
        tool = tool_cls(*args)
        self.add(tool)

1.3 依赖注入系统

工具的依赖通过类型注解自动注入：

# 运行时依赖
tool_deps = {
    KimiToolset: toolset,
    Runtime: runtime,
    Config: runtime.config,
    BuiltinSystemPromptArgs: runtime.builtin_args,
    Session: runtime.session,
    DenwaRenji: runtime.denwa_renji,
    Approval: runtime.approval,
    LaborMarket: runtime.labor_market,
    Environment: runtime.environment,
}

toolset.load_tools(tools, tool_deps)

1.4 工具调用流程

用户请求 → Agent → LLM → 工具调用 → Toolset.handle() 
         → 工具实例 → 执行 → 返回结果 → Agent → 用户

核心实现：

def handle(self, tool_call: ToolCall) -> HandleResult:
    token = current_tool_call.set(tool_call)
    try:
        tool = self._tool_dict[tool_call.function.name]
        
        # 解析参数
        arguments: JsonType = json.loads(tool_call.function.arguments or "{}")
        
        # 异步调用
        async def _call():
            ret = await tool.call(arguments)
            return ToolResult(tool_call_id=tool_call.id, return_value=ret)
        
        return asyncio.create_task(_call())
    finally:
        current_tool_call.reset(token)

1.5 工具分类

文件工具

ReadFile: 读取文件内容
WriteFile: 写入文件
StrReplaceFile: 字符串替换文件内容
Glob: 文件模式匹配
Grep: 文本搜索
ReadMediaFile: 读取媒体文件

Web 工具

SearchWeb: 网络搜索
FetchURL: 抓取网页内容

多 Agent 工具

Task: 创建子任务
CreateSubagent: 创建子 Agent

Shell 工具

Shell: 执行 Shell 命令

工具辅助

Think: 内部思考
SendDMail: 发送 D-Mail
SetTodoList: 任务列表管理

2. 技能系统

2.1 技能定义

技能是可复用的能力组合，每个技能是一个包含 SKILL.md 的目录：

skills/
├── kimi-cli-help/
│   └── SKILL.md
└── skill-creator/
    └── SKILL.md

SKILL.md 格式：

---
name: kimi-cli-help
description: 回答 Kimi CLI 使用、配置和故障排除问题
---

# Kimi Code CLI Help

技能说明和使用指南...

## Strategy

使用策略说明...

2.2 技能发现机制

技能从多个层级发现（src/kimi_cli/skill/__init__.py）：

# 优先级顺序
1. 内置技能: src/kimi_cli/skills/
2. 用户技能: ~/.config/agents/skills/
3. 用户技能: ~/.agents/skills/
4. 用户技能: ~/.kimi/skills/
5. 项目技能: .agents/skills/

发现实现：

async def resolve_skills_roots(
    work_dir: KaosPath,
    *,
    skills_dir_override: KaosPath | None = None,
) -> list[KaosPath]:
    roots: list[KaosPath] = []
    if _supports_builtin_skills():
        roots.append(KaosPath.unsafe_from_local_path(get_builtin_skills_dir()))
    if skills_dir_override is not None:
        roots.append(skills_dir_override)
        return roots
    if user_dir := await find_user_skills_dir():
        roots.append(user_dir)
    if project_dir := await find_project_skills_dir(work_dir):
        roots.append(project_dir)
    return roots

2.3 技能类型

标准技能

标准的文本指令技能，通过斜杠命令调用：

# 在 KimiSoul 中注册
def _build_slash_commands(self) -> list[SlashCommand[Any]]:
    for skill in self._runtime.skills.values():
        name = f"skill:{skill.name}"
        commands.append(
            SlashCommand(
                name=name,
                func=self._make_skill_runner(skill),
                description=skill.description or "",
            )
        )
    return commands

Flow 技能

基于流程图的技能，使用 Mermaid 或 D2 定义：

---
name: code-review
type: flow
description: 代码审查流程
---

```mermaid
graph TD
    BEGIN --> READ
    READ --> ANALYZE
    ANALYZE --> DECISION
    DECISION -->|有问题| FIX
    DECISION -->|无问题| END
    FIX --> VERIFY
    VERIFY --> END


Flow 技能解析：

```python
def _parse_flow_from_skill(content: str) -> Flow:
    for lang, code in _iter_fenced_codeblocks(content):
        if lang == "mermaid":
            return _parse_flow_block(parse_mermaid_flowchart, code)
        if lang == "d2":
            return _parse_flow_block(parse_d2_flowchart, code)
    raise ValueError("Flow skills require a mermaid or d2 code block in SKILL.md.")

2.4 技能调用

技能通过斜杠命令调用，如 /skill:kimi-cli-help：

def _make_skill_runner(self, skill: Skill) -> Callable[[KimiSoul, str], None | Awaitable[None]]:
    async def _run_skill(soul: KimiSoul, args: str) -> None:
        skill_text = await read_skill_text(_skill)
        if skill_text is None:
            wire_send(TextPart(text=f'Failed to load skill "/{SKILL_COMMAND_PREFIX}{_skill.name}".'))
            return
        extra = args.strip()
        if extra:
            skill_text = f"{skill_text}\n\nUser request:\n{extra}"
        await soul._turn(Message(role="user", content=skill_text))
    return _run_skill

2.5 技能在系统提示词中的呈现

技能信息会被注入到系统提示词中：

BuiltinSystemPromptArgs(
    KIMI_SKILLS: str = """- kimi-cli-help
  - Path: /path/to/skills/kimi-cli-help/SKILL.md
  - Description: 回答 Kimi CLI 使用、配置和故障排除问题
- skill-creator
  - Path: /path/to/skills/skill-creator/SKILL.md
  - Description: 创建新的技能"""
)

系统提示词模板 (system.md):

# Skills

Skills are reusable, composable capabilities that enhance your abilities.

## Available skills

${KIMI_SKILLS}

## How to use skills

Identify the skills that are likely to be useful for the tasks you are currently working on, read the `SKILL.md` file for detailed instructions, guidelines, scripts and more.

3. MCP (Model Context Protocol) 工具

3.1 MCP 概述

MCP 是一个开放协议，允许 AI 模型安全地与外部工具和数据源交互。Kimi-CLI 支持连接 MCP 服务器以扩展 AI 能力。

3.2 MCP 服务器管理

添加服务器

# HTTP 服务器
kimi mcp add --transport http context7 https://mcp.context7.com/mcp \
  --header "CONTEXT7_API_KEY: your-key"

# OAuth 认证
kimi mcp add --transport http --auth oauth linear https://mcp.linear.app/mcp

# Stdio 服务器
kimi mcp add --transport stdio chrome-devtools -- npx chrome-devtools-mcp@latest

列出服务器

kimi mcp list

删除服务器

kimi mcp remove context7

认证

kimi mcp auth linear

3.3 MCP 工具加载

async def load_mcp_tools(
    self, 
    mcp_configs: list[MCPConfig], 
    runtime: Runtime, 
    in_background: bool = True
) -> None:
    import fastmcp
    
    # 创建客户端
    for server_name, server_config in mcp_config.mcpServers.items():
        client = fastmcp.Client(MCPConfig(mcpServers={server_name: server_config}))
        self._mcp_servers[server_name] = MCPServerInfo(
            status="pending", 
            client=client, 
            tools=[]
        )
    
    # 连接并加载工具
    async def _connect():
        for server_name, server_info in self._mcp_servers.items():
            async with server_info.client as client:
                for tool in await client.list_tools():
                    mcp_tool = MCPTool(server_name, tool, client, runtime=runtime)
                    server_info.tools.append(mcp_tool)
                    self.add(mcp_tool)
    
    if in_background:
        self._mcp_loading_task = asyncio.create_task(_connect())
    else:
        await _connect()

3.4 MCP 工具实现

class MCPTool[T: ClientTransport](CallableTool):
    def __init__(
        self,
        server_name: str,
        mcp_tool: mcp.Tool,
        client: fastmcp.Client[T],
        *,
        runtime: Runtime,
        **kwargs: Any,
    ):
        super().__init__(
            name=mcp_tool.name,
            description=(
                f"This is an MCP tool from MCP server `{server_name}`.\n\n"
                f"{mcp_tool.description or 'No description provided.'}"
            ),
            parameters=mcp_tool.inputSchema,
            **kwargs,
        )
        self._mcp_tool = mcp_tool
        self._client = client
        self._runtime = runtime
        self._timeout = timedelta(
            milliseconds=runtime.config.mcp.client.tool_call_timeout_ms
        )
    
    async def __call__(self, *args: Any, **kwargs: Any) -> ToolReturnValue:
        # 请求批准
        description = f"Call MCP tool `{self._mcp_tool.name}`."
        if not await self._runtime.approval.request(
            self.name, 
            self._action_name, 
            description
        ):
            return ToolRejectedError()
        
        # 调用 MCP 工具
        async with self._client as client:
            result = await client.call_tool(
                self._mcp_tool.name,
                kwargs,
                timeout=self._timeout,
                raise_on_error=False,
            )
            return convert_mcp_tool_result(result)

3.5 ACP-MCP 集成

ACP (Agent Client Protocol) 客户端可以传递 MCP 配置：

def acp_mcp_servers_to_mcp_config(mcp_servers: list[MCPServer]) -> MCPConfig:
    return MCPConfig.model_validate(
        {"mcpServers": {
            server.name: _convert_acp_mcp_server(server) 
            for server in mcp_servers
        }}
    )

def _convert_acp_mcp_server(server: MCPServer) -> dict[str, Any]:
    match server:
        case acp.schema.HttpMcpServer():
            return {
                "url": server.url,
                "transport": "http",
                "headers": {header.name: header.value for header in server.headers},
            }
        case acp.schema.SseMcpServer():
            return {
                "url": server.url,
                "transport": "sse",
                "headers": {header.name: header.value for header in server.headers},
            }
        case acp.schema.McpServerStdio():
            return {
                "command": server.command,
                "args": server.args,
                "env": {item.name: item.value for item in server.env},
                "transport": "stdio",
            }

4. 扩展性设计

4.1 工具扩展

开发者可以创建自定义工具：

创建工具类，继承 CallableTool2
定义参数模型
实现 __call__ 方法
在 agent.yaml 中注册

# my_tool.py
from kosong.tooling import CallableTool2, ToolReturnValue
from pydantic import BaseModel

class Params(BaseModel):
    text: str

class MyTool(CallableTool2[Params]):
    name: str = "MyTool"
    description: str = "我的自定义工具"
    params: type[Params] = Params
    
    async def __call__(self, params: Params) -> ToolReturnValue:
        return ToolOk(output=[TextPart(text=params.text)])

4.2 技能扩展

开发者可以创建自定义技能：

在技能目录创建 SKILL.md
编写技能说明
放到 ~/.config/agents/skills/ 或项目 .agents/skills/

4.3 MCP 服务器扩展

开发者可以创建 MCP 服务器：

实现 MCP 协议
使用 fastmcp 框架
通过 kimi mcp add 添加

5. 配置文件

Agent 配置

# agent.yaml
version: 1
agent:
  name: "my-agent"
  system_prompt_path: ./system.md
  system_prompt_args:
    ROLE_ADDITIONAL: "专注于代码审查"
  tools:
    - "kimi_cli.tools.file:ReadFile"
    - "kimi_cli.tools.web:SearchWeb"
  exclude_tools:
    - "kimi_cli.tools.shell:Shell"
  subagents:
    coder:
      path: ./sub.yaml
      description: "擅长软件工程任务"

MCP 配置

# config.toml
[mcp.client]
tool_call_timeout_ms = 60000

[mcp.servers]
[mcp.servers.context7]
url = "https://mcp.context7.com/mcp"
transport = "http"
headers = { CONTEXT7_API_KEY = "your-key" }

[mcp.servers.chrome-devtools]
command = "npx"
args = ["-y", "chrome-devtools-mcp@latest"]
transport = "stdio"