Kimi SDK 使用指南：用 Python 快速构建基于 Kimi API 的 Agent 工作流

Kimi SDK 使用指南：用 Python 快速构建基于 Kimi API 的 Agent 工作流

【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cli

导读

Kimi SDK 是 kimi-cli 仓库中提供的轻量级 Python 封装，它基于 monorepo 内的kosongLLM 抽象层，为开发者提供了一条直达 Kimi API（Moonshot 平台）的便捷通道：你只需几行代码即可完成对话补全、流式输出、视频文件上传与工具调用，并能进一步搭建完整的 Agent 循环。读完本文，你将掌握kimi-sdk的安装方式、四个核心 API（Kimi、generate、step、SimpleToolset）的用法，以及它们底层的实现原理与可复用的实战模式。

Kimi SDK 是什么

Kimi SDK（包定义）定位为 "A lightweight Python SDK for the Kimi API"，当前版本 0.2.1，要求 Python 3.12 及以上。它本身是一个极薄的封装层：真正的 LLM 抽象逻辑（消息结构、异步工具编排、可插拔的聊天 Provider）全部位于同仓库的kosong包中，kimi-sdk通过依赖kosong>=0.37.0并从中精选导出与 Kimi API 直接相关的能力。

这一点在 包入口 中体现得很清楚：kimi_sdk的__init__.py不做任何业务实现，而是从kosong的chat_provider、message、tooling等模块批量导出符号，并维护一份__all__白名单。你从kimi_sdk导入的每个名字，其实现都可以在 kosong 源码 中找到对应定义。

核心能力可归纳为三条：

• generate：发起一次补全，把流式返回的 message parts 合并成完整的Message，并附上可选的TokenUsage用量统计；
• step：在generate之上叠加工具分发（Tool/Toolset/SimpleToolset），返回带工具输出结果的StepResult；
• 消息与工具抽象：Message、各类ContentPart（文本、思考、图片、音频、视频）、ToolCall等数据结构统一在这里定义。

安装与项目初始化

官方推荐使用uv作为包管理器（uv 也承担本仓库的构建与开发工作流）：

uv init --python 3.12 # or higher uv add kimi-sdk

安装后即可在代码中导入：

from kimi_sdk import Kimi, Message, generate

运行环境要求

• Python >= 3.12（见 pyproject.toml 的requires-python字段）；
• 依赖kosong>=0.37.0，后者内部依赖openaiSDK 与httpx完成 HTTP 通信；
• 调用 API 前需准备 Moonshot 平台的 API Key（可通过环境变量注入，见下文"环境变量"一节）。

第一个示例：简单的对话补全

这是 README 中最基础也最核心的用法——创建KimiProvider，构造历史消息，然后调用generate获取回复：

import asyncio from kimi_sdk import Kimi, Message, generate async def main() -> None: kimi = Kimi( base_url="https://api.moonshot.ai/v1", api_key="your_kimi_api_key_here", model="kimi-k2-turbo-preview", ) history = [ Message(role="user", content="Who are you?"), ] result = await generate( chat_provider=kimi, system_prompt="You are a helpful assistant.", tools=[], history=history, ) print(result.message) print(result.usage) asyncio.run(main())

关键参数说明

generate返回GenerateResult，其中message是模型生成的完整消息（可直接print查看文本），usage是TokenUsage用量对象。从 GenerateResult 定义 可以看到它还携带id（消息 ID）与trace_id（响应的x-trace-id请求头，便于排查问题）。

流式输出：逐块接收消息

generate默认走流式通道，你可以通过on_message_part回调实时拿到每一个到达的 message part，从而实现打字机式的输出效果：

import asyncio from kimi_sdk import Kimi, Message, StreamedMessagePart, generate async def main() -> None: kimi = Kimi( base_url="https://api.moonshot.ai/v1", api_key="your_kimi_api_key_here", model="kimi-k2-turbo-preview", ) history = [ Message(role="user", content="Who are you?"), ] def output(message_part: StreamedMessagePart) -> None: print(message_part) result = await generate( chat_provider=kimi, system_prompt="You are a helpful assistant.", tools=[], history=history, on_message_part=output, ) print(result.message) print(result.usage) asyncio.run(main())

流式 part 如何被合并

在 generate 实现 中，generate先调用chat_provider.generate(...)拿到一个异步流，然后逐块迭代：

• 每个 part 先通过on_message_part回调以深拷贝形式暴露给调用方（避免外部修改污染内部状态）；
• 如果前一个 part 尚未完成，会尝试用pending_part.merge_in_place(part)把新的分片合并进去（例如被切分的文本增量），合并不了才推入消息缓冲区；
• 流结束时把所有待处理的 part 写入最终Message，并做异常兜底：如果响应内容为空，或只有ThinkPart（思考内容）却没有可见文本和工具调用，则抛出APIEmptyResponseError——后者通常意味着流被中断或输出 token 预算在推理阶段耗尽。

因此result.message始终是合并完整、可直接使用的消息，而on_message_part只用于实时展示。

上传视频：把本地文件变成消息内容

Kimi SDK 支持把视频作为多模态输入送入对话。核心是kimi.files.upload_video(...)，它走 KimiFiles 实现：

import asyncio from pathlib import Path from kimi_sdk import Kimi, Message, TextPart, generate async def main() -> None: kimi = Kimi( base_url="https://api.moonshot.ai/v1", api_key="your_kimi_api_key_here", model="kimi-k2-turbo-preview", ) video_path = Path("demo.mp4") video_part = await kimi.files.upload_video( data=video_path.read_bytes(), mime_type="video/mp4", ) history = [ Message( role="user", content=[ TextPart(text="Please describe this video."), video_part, ], ), ] result = await generate( chat_provider=kimi, system_prompt="You are a helpful assistant.", tools=[], history=history, ) print(result.message) print(result.usage) asyncio.run(main())

upload_video 的底层行为

• mime_type必须以video/开头，否则抛出ChatProviderError；
• SDK 通过底层 OpenAI 兼容客户端向/files接口发起multipart/form-data上传，purpose固定为"video"，文件名由mimetypes根据 MIME 类型推断；
• 上传成功后返回一个VideoURLPart，其内部 URL 形如ms://{file_id}，可直接作为Message.content的一个元素参与后续补全；
• 消息内容因此支持多种ContentPart组合：除了TextPart与VideoURLPart，包入口 还导出了ThinkPart、ImageURLPart、AudioURLPart等，可用于构建富媒体对话。

工具调用：基于 step 的 Agent 单步执行

当模型需要调用外部函数时，使用step而不是generate。step在generate之上自动完成工具调用的分发，一个经典的整数加法工具示例如下：

import asyncio from pydantic import BaseModel from kimi_sdk import CallableTool2, Kimi, Message, SimpleToolset, StepResult, ToolOk, ToolReturnValue, step class AddToolParams(BaseModel): a: int b: int class AddTool(CallableTool2[AddToolParams]): name: str = "add" description: str = "Add two integers." params: type[AddToolParams] = AddToolParams async def __call__(self, params: AddToolParams) -> ToolReturnValue: return ToolOk(output=str(params.a + params.b)) async def main() -> None: kimi = Kimi( base_url="https://api.moonshot.ai/v1", api_key="your_kimi_api_key_here", model="kimi-k2-turbo-preview", ) toolset = SimpleToolset() toolset += AddTool() history = [ Message(role="user", content="Please add 2 and 3 with the add tool."), ] result: StepResult = await step( chat_provider=kimi, system_prompt="You are a precise math tutor.", toolset=toolset, history=history, ) print(result.message) print(await result.tool_results()) asyncio.run(main())

定义一个工具的三要素

以CallableTool2[T]为基类、T为 Pydantic 参数模型：

• name：工具名，会作为 function calling 的name发给模型；
• description：工具说明，帮助模型决定何时调用；
• params：参数 schema（Pydantic 模型类）；
• __call__：真正的执行逻辑，返回ToolReturnValue。返回ToolOk(output=...)表示成功，另有ToolError表示失败，两者定义在 tooling 模块 中。

SimpleToolset支持+=语法批量注册工具，toolset.tools会在请求时转换为 OpenAI 兼容的工具参数格式。

step 与 StepResult 的行为契约

从 step 源码 可以确认以下细节：

• step每调用一次只让模型生成一轮，工具调用由toolset.handle(tool_call)立即分发执行；执行结果包装在ToolResultFuture中，可通过await result.tool_results()统一取回；
• StepResult暴露id、message、usage、tool_calls与tool_results()，其中tool_calls列出本轮模型请求的全部工具调用；
• step不会修改传入的 history——是否把本轮消息与工具结果追加回历史，由调用方决定，这让 Agent 循环的状态管理完全透明可控；
• 若生成过程中出现ChatProviderError或任务取消，step会取消所有未完成的工具 future，避免后台任务悬挂；
• 支持的异常类型包括APIConnectionError、APITimeoutError、APIStatusError（4xx/5xx）、APIEmptyResponseError及统一的ChatProviderError基类。

实战：完整的 Agent 循环

将step与手动维护的history组合，即可写出一个可交互的 Agent 主循环（该示例来自 kimi_sdk 包入口文档 并在此展开）：

import asyncio from kimi_sdk import Kimi, Message, SimpleToolset, StepResult, ToolResult, step def tool_result_to_message(result: ToolResult) -> Message: return Message( role="tool", tool_call_id=result.tool_call_id, content=result.return_value.output, ) async def agent_loop() -> None: kimi = Kimi( base_url="https://api.moonshot.ai/v1", api_key="your_kimi_api_key_here", model="kimi-k2-turbo-preview", ) toolset = SimpleToolset() # toolset += YourTool() # 注册你的工具 history: list[Message] = [] system_prompt = "You are a helpful assistant." while True: user_input = input("You: ").strip() if not user_input: continue if user_input.lower() in {"exit", "quit"}: break history.append(Message(role="user", content=user_input)) while True: result: StepResult = await step( chat_provider=kimi, system_prompt=system_prompt, toolset=toolset, history=history, ) history.append(result.message) tool_results = await result.tool_results() for tool_result in tool_results: history.append(tool_result_to_message(tool_result)) if text := result.message.extract_text(): print("Assistant:", text) if not result.tool_calls: break asyncio.run(agent_loop())

这个循环揭示了 Agent 的标准运转模式：

1. 读取用户输入，追加为user消息；
2. 内层循环反复执行step：每轮把模型消息写入历史、取回工具结果并转成role="tool"的消息回填历史；
3. Message.extract_text()用于抽取当前轮次的可见文本并打印；
4. 当result.tool_calls为空（模型不再调用工具）时退出内层循环，回到等待用户输入。

role="tool"的消息通过tool_call_id与模型发出的ToolCall一一对应，这正是 OpenAI 兼容 function calling 协议所要求的闭环格式。

环境变量

Kimi SDK 支持两个环境变量，其读取逻辑在 Kimi 构造器 中：

注意优先级：显式传入的构造参数优先于环境变量。例如Kimi(model=..., api_key="sk-xxx")会忽略KIMI_API_KEY；而Kimi(model=...)则会读取KIMI_API_KEY。

深入：Kimi Provider 的高级能力

除了 README 示例，Kimi 类实现 还提供了不少值得在生产中使用的进阶特性：

生成参数（GenerationKwargs）

支持max_completion_tokens（旧别名max_tokens会自动归一化）、temperature、top_p、n、presence_penalty、frequency_penalty、stop、prompt_cache_key、reasoning_effort与extra_body等，通过with_generation_kwargs以不可变副本方式应用：

kimi = kimi.with_generation_kwargs(temperature=0, max_completion_tokens=1000)

思考模式与 preserved thinking

with_thinking(effort)接受ThinkingEffort（如"off"），底层通过extra_body.thinking.type控制是否启用思考；Moonshot 特有的thinking.keep（如"all"）用于保留推理内容，由with_extra_body按字段合并，不会覆盖已设置的thinking.type。

内置函数与 schema 兼容

• 工具名以$开头的工具会被映射为 Kimi 内置函数（builtin_function），无需提供 description 和 parameters；
• 自动对工具参数 schema 做ensure_property_types归一化，修复部分 MCP 服务器产出的"缺省type的嵌套属性"导致的 400 错误。

用量与可观测性

TokenUsage区分input_other（非缓存输入）、output（输出）与input_cache_read（缓存命中输入）；KimiStreamedMessage会兼容处理 Moonshot 与 OpenAI 两种 usage 字段格式。StepResult.trace_id与generate的on_trace_id回调可拿到响应的x-trace-id，方便链路追踪。

测试验证

仓库为 SDK 提供了冒烟测试 tests/test_smoke.py，使用httpx.MockTransport拦截请求并返回模拟的 chat completion 响应，验证了：

• generate会向/v1/chat/completions发起请求；
• 非流式模式下（stream=False）result.message.extract_text()能正确取出"Hello"；
• result.usage.input_other == 10、result.usage.output == 5的用量解析逻辑正确。

这个测试模式非常实用：通过注入自定义http_client（httpx.AsyncClient），你可以在不真正调用 Kimi API 的情况下对自家 Agent 逻辑做单元测试。

版本演进

从 CHANGELOG 可以看到 SDK 的发展脉络：0.1.0 首次发布；0.2.0 导出KimiFiles以支持视频文件上传；0.2.1 放宽kosong依赖上限以兼容 0.40.x。如果你在本仓库中开发，可以留意kosong的版本约束与kimi-sdk的导出面随版本逐步扩充。

小结

Kimi SDK 用极薄的 API 面把"连接 Kimi API + 构建 Agent 工作流"这件事做到了开箱即用：generate负责对话与流式合并，step负责工具分发，SimpleToolset+CallableTool2负责工具定义，Message与ContentPart负责统一的多模态消息模型。结合底层kosong源码阅读，你既能快速上手，也能理解流式合并、工具 future 编排、异常兜底等内部机制，从而在自己的 Python 项目中搭建稳定、可测试的 Agent 应用。

【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cli