cmdc-sdk 0.1.0

CMDC Agent Kernel Python SDK — connect to cmdc_gateway via HTTP + SSE + WebSocket

cmdc-python

CMDC Agent Kernel Python SDK — 通过 HTTP + SSE + WebSocket 连接 cmdc_gateway，在 Python 中使用 CMDC Agent 能力。

---

安装

pip install cmdc

或从源码安装（开发模式）：

cd python
pip install -e ".[dev]"

依赖要求：

• Python 3.10+
• httpx >= 0.27
• httpx-sse >= 0.4
• websockets >= 12.0

---

快速开始（5 分钟）

1. 启动 cmdc_gateway

cd cmdc_gateway
CMDC_API_KEYS=sk-abc123 mix run --no-halt

2. 最简对话

from cmdc import CMDC

client = CMDC(base_url="http://localhost:4000", api_key="sk-abc123")

with client.session(model="deepseek:deepseek-chat") as session:
    for event in session.prompt("帮我写一个快速排序算法"):
        if event.type == "message_delta":
            print(event.delta, end="", flush=True)
    print()  # 换行

3. 流式输出（完整示例）

from cmdc import CMDC

with CMDC(base_url="http://localhost:4000", api_key="sk-abc123") as client:
    with client.session(
        model="deepseek:deepseek-chat",
        system_prompt="你是一个专业的 Python 编程助手",
        tools=["CMDC.Tool.Shell", "CMDC.Tool.ReadFile"],
        max_turns=20,
    ) as session:
        full_reply = ""
        for event in session.prompt("分析当前目录的 Python 文件"):
            match event.type:
                case "message_delta":
                    print(event.delta, end="", flush=True)
                    full_reply += event.delta or ""
                case "tool_execution_start":
                    print(f"\n[工具] {event.data['toolName']} 开始执行...")
                case "tool_execution_end":
                    status = event.data.get("status", "?")
                    print(f"[工具] 执行完成 ({status})")
                case "agent_end":
                    tokens = event.token_usage
                    if tokens:
                        print(f"\n[完成] 共消耗 {tokens.total_tokens} tokens")

---

API 参考

CMDC 主客户端

client = CMDC(
    base_url="http://localhost:4000",  # Gateway 地址
    api_key="sk-abc123",               # API Key
    timeout=30.0,                      # 请求超时（秒）
    connect_timeout=10.0,              # 连接超时（秒）
)

方法

方法	说明
create_session(model, **kwargs)	创建 Session，返回 Session 实例
session(model, **kwargs)	Context Manager，自动 stop
create_session_async(model, **kwargs)	异步版 create_session
asession(model, **kwargs)	异步 Context Manager
health()	检查 Gateway 健康状态
close() / aclose()	关闭 HTTP 连接池

SessionOptions 字段

from cmdc import SessionOptions

opts = SessionOptions(
    model="deepseek:deepseek-chat",   # 必填
    session_id="my-session-001",       # 自定义 ID（可选）
    system_prompt="你是助手",           # 系统提示词
    working_dir="/home/user/project",  # 工具工作目录
    tools=["CMDC.Tool.Shell"],         # 启用的工具模块
    plugins=["CMDC.Plugin.Builtin.SecurityGuard"],  # 插件
    blueprint="CMDC.Blueprint.Base",   # 蓝图模块
    max_turns=50,                      # 最大轮次
    max_tokens=4096,                   # 最大输出 token
    skills_dirs=["/home/user/skills"], # Skill 目录
    provider_opts={"temperature": 0.7},# Provider 额外参数
)
session = client.create_session(options=opts)

---

Session 操作

发送消息

# 方式一：迭代事件流
for event in session.prompt("你好"):
    if event.type == "message_delta":
        print(event.delta, end="")

# 方式二：阻塞等待完整回复
reply = session.prompt_sync("简单回答：1+1=?")
print(reply)  # "2"

# 方式三：仅投递，不监听
response = session.send_prompt("开始执行任务")
print(response.request_id)  # 16 位 hex

控制操作

session.approve("apr_x7y8z9")          # 审批通过
session.reject("apr_x7y8z9")           # 审批拒绝
session.respond("ask_001", "快速排序") # 回答 Agent 提问
session.stop()                          # 停止 Agent

查询

info = session.get_info()       # SessionInfo
stats = session.get_stats()     # Stats（含 token 用量）
messages = session.get_messages()  # List[Message]

---

事件类型（15 种）

类型	说明	常用字段
agent_start	Agent 开始处理	—
agent_end	处理完成	event.last_message, event.token_usage
agent_abort	Agent 被中止	event.data["reason"]
prompt_received	确认收到 prompt	event.data["text"]
message_start	LLM 开始生成	—
message_delta	流式文本片段	event.delta
thinking_start	思考链开始	—
thinking_delta	思考文本片段	event.data["delta"]
tool_calls	本轮工具数量	event.data["count"]
tool_execution_start	工具开始执行	event.data["toolName"], event.data["args"]
tool_execution_end	工具执行完成	event.data["status"], event.data["result"]
approval_required	需要人工审批	event.approval_id, event.data["toolName"]
approval_resolved	审批已决定	event.data["status"]
ask_user	Agent 向用户提问	event.ask_ref, event.question
error	运行时错误	event.data["reason"]

---

HITL 审批（HumanApproval）

from cmdc import CMDC, AutoApprove, AutoReject, PromptApproval

# 自动批准所有工具
for event in session.prompt("执行清理任务", approval_policy="auto_approve"):
    ...

# 终端手动审批
for event in session.prompt("执行清理任务", approval_policy="prompt"):
    ...

# 自定义逻辑：只允许非 Shell 工具
policy = lambda e: e.data.get("toolName") != "Shell"
for event in session.prompt("执行任务", approval_policy=policy):
    ...

---

本地工具注册

from cmdc import LocalTool

def search_db(args: dict) -> str:
    query = args.get("query", "")
    # 本地数据库查询逻辑
    return f"查询结果: {query}"

tool = LocalTool(
    name="search_database",
    description="在本地数据库中搜索信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词"}
        },
        "required": ["query"],
    },
    callback=search_db,
    timeout=10.0,
)

# 向 Gateway 注册（需要本地 HTTP 服务监听 callback_url）
session.register_tool(tool, callback_url="http://my-service:8080/tools/search_database")

---

WebSocket 双向连接

# 同步版本
with session.connect_ws() as ws:
    ws.send_prompt("执行危险操作")
    for event in ws.iter_events():
        if event.type == "approval_required":
            print(f"需要审批: {event.data['toolName']}")
            ws.approve(event.approval_id)
        elif event.type == "message_delta":
            print(event.delta, end="")
        elif event.type == "agent_end":
            break

# 异步版本
async with session.connect_ws() as ws:
    await ws.send_prompt("执行危险操作")
    async for event in ws.iter_events():
        if event.type == "approval_required":
            await ws.approve(event.approval_id)
        elif event.type == "agent_end":
            break

---

异步 API

import asyncio
from cmdc import CMDC

async def main():
    async with CMDC(base_url="http://localhost:4000", api_key="sk-abc123") as client:
        async with client.asession(model="deepseek:deepseek-chat") as session:
            gen = await session.aprompt("你好")
            async for event in gen:
                if event.type == "message_delta":
                    print(event.delta, end="", flush=True)

asyncio.run(main())

---

错误处理

from cmdc import (
    CMDCError,
    AuthError,
    SessionNotFoundError,
    RateLimitError,
    SessionCreateError,
)

try:
    session = client.create_session(model="unknown:model")
except AuthError:
    print("API Key 无效")
except SessionCreateError as e:
    print(f"创建失败: {e.message}")
except RateLimitError as e:
    print(f"超出限流，{e.retry_after}s 后重试")
except CMDCError as e:
    print(f"其他错误 [{e.status_code}]: {e.message}")

---

运行测试

cd python
pip install -e ".[dev]"
pytest tests/ -v

---

环境变量

变量	说明
CMDC_BASE_URL	Gateway 地址，默认 http://localhost:4000
CMDC_API_KEY	API Key

快速配置：

import os
from cmdc import CMDC

client = CMDC(
    base_url=os.environ.get("CMDC_BASE_URL", "http://localhost:4000"),
    api_key=os.environ["CMDC_API_KEY"],
)