DuMate Python SDK

面向 DuMate 老版 /api/v1 的 Python 客户端。Python 3.10+，提供同步 DuMate 和异步 AsyncDuMate。当前版本 0.3.0，为独立开发的 SDK。

应用层快速接入与前端数据协议见 APPLICATIONS.md。提供 skills.ensure、conversations.create/history/stream 和 notebook.stream。

安装

pip install ./dmt_agent_api-0.3.0-py3-none-any.whl

源码安装：pip install .。配置环境变量 DUMATE_API_KEY；可选 DUMATE_BASE_URL，默认 https://api.dumate.cn/api/v1。Key 仅放在后端。

一次调用，等待最终结果

from dumate import DuMate

with DuMate() as client:
    agent = client.agents.create(name="我的助手")
    session = client.sessions.create(agent_id=agent["id"])
    result = client.runs.run_and_wait(
        session_id=session["id"],
        message="介绍你能完成的工作",
        timeout=180,
    )
    print(result.output_text)

runs.run_and_wait 在客户端发送一次消息并轮询历史，关联本次用户消息与最终回答；没有对应的服务端 /runs 接口。适合网页的一键任务按钮：后端执行调用，完成后返回 output_text，前端展示结果即可。较长任务应由应用队列执行并提供任务状态查询。

同一会话串行发送，不同会话可以并发。超时仅停止本地等待，服务端任务可能仍在执行；可查询历史或显式 sessions.interrupt。timeout 限制轮询等待，单次网络请求还受客户端 timeout 约束。默认等待时间无限，建议显式设置。

流式事件

with client.sessions.stream(session_id, message="帮我分析这个问题") as events:
    for event in events:
        if event.type == "message.part.updated":
            part = event.properties.get("part", {})
            print(part.get("type"), part.get("text", ""))
        elif event.type == "session.idle" or (
            event.type == "session.status"
            and event.properties.get("status", {}).get("type") == "idle"
        ):
            break

流建立后才发送消息。事件保留原始 type、properties 和 raw，未知事件也会返回；SSE id、event 字段独立保留。应用处理规则：

SDK 提供事件解析，不包含网页组件。断流抛出 StreamInterruptedError；应用先恢复 sessions.history，再重新订阅，不能自动重发原 query。SDK 不承诺事件重放或 Last-Event-ID 恢复。完整事件字段以服务端实际返回为准。

Skill

ZIP 中包含技能目录和 SKILL.md，文件带 YAML name、description。

skill = client.skills.upload("my-skill.zip")
mount = client.skills.mount(skill["id"])
session = client.sessions.create(agent_id=agent_id, skills=[mount])

updated = client.skills.update_zip(skill["id"], "my-skill-v2.zip")
new_mount = client.skills.mount(skill["id"])
client.sessions.update(session["id"], skills=[new_mount])

挂载使用 skillID、releaseID、type；mount() 从详情取得 releaseId，不能把展示版本号当作发布 ID。URL 导入用 skills.create(url=...)，URL 更新用 skills.update(skill_id, url=...)。

Vault、凭据和 MCP

import os

vault = client.vaults.create(name="业务系统")
credentials = client.vaults.credentials(vault["id"])
credential = credentials.create(
    name="业务访问令牌",
    auth={
        "type": "environment_variable",
        "secretName": "BUSINESS_TOKEN",
        "secretValue": os.environ["BUSINESS_TOKEN"],
        "networking": {
            "type": "limited",
            "allowedHosts": ["api.your-company.com"],
        },
    },
)
session = client.sessions.create(agent_id=agent_id, vault_ids=[vault["id"]])

auth 原样传递平台的 environment_variable、static_bearer 或 mcp_oauth 结构，敏感值不要记录到日志。MCP 配置通过 mcp_servers 传入 Agent 或 Session，例如 [{"type":"url","name":"business","url":"https://your-company.com/mcp"}]；会话配置 MCP 时同时提供对应 vault_ids。各 MCP 的认证字段须按官方文档及服务商要求填写；SDK 不代办 OAuth 授权。平台未提供凭据更新接口，轮换需创建新凭据并迁移使用后删除旧凭据。

文件与产物

uploaded = client.files.upload("input.pdf", session_id=session_id)
result = client.runs.run_and_wait(
    session_id=session_id, message="总结附件", file_ids=[uploaded["fileID"]], timeout=300,
)
client.files.download(uploaded["fileID"], "downloaded.pdf")

mapping = client.artifacts.path_map(session_id)
for path, artifact in mapping.items():
    if artifact.get("uploadStatus") == "ready":
        client.artifacts.download(session_id, artifact["artifactID"], "output.bin")
        break

path-map 是整个会话的产物集合，不能全部当作本轮产物。按本轮回答中的文件路径、工具结果或执行前后的映射差异关联。output.bin 应由业务层按选中的文件设置名称；不要直接信任远端路径。

下载返回落盘信息，默认禁止覆盖，支持 max_bytes。文件先写临时文件，成功后提交；校验响应长度并计算 SHA-256。产物必须 ready；存储返回 403 时刷新一次 URL。存储下载不携带 DuMate Key。URL 校验阻止明显私网地址及不安全跳转；生产环境仍需网络出口策略防止 DNS 重绑定。

预览由应用后端代理文件，前端按 MIME 展示图片、音视频、PDF；HTML 应隔离在 sandbox iframe 中，未知类型提供下载。签名 URL 有时效，不应长期保存为产物身份。

异步

import asyncio
from dumate import AsyncDuMate

async def main():
    async with AsyncDuMate() as client:
        session = await client.sessions.create(agent_id="你的 Agent ID")
        result = await client.runs.run_and_wait(
            session_id=session["id"], message="你好", timeout=180,
        )
        print(result.output_text)

asyncio.run(main())

异步资源方法使用 await；列表迭代用 async for item in client.agents.iter()；流用 async with client.sessions.stream(...) as events 和 async for。文件落盘及 DNS 检查仍包含短同步操作，大文件应用可用专门下载工作进程。

错误与运行边界

• GET 对网络错误、429 和部分 5xx 做有限重试，默认 2 次；写操作不自动重试。
• 写请求连接失败时 ConnectionError.outcome_unknown=True，先核对服务端状态，避免重复创建或执行。
• APIError 包含状态码及请求 ID；异常文本不包含响应正文或 Key。应用不要直接打印凭据查询返回值。
• RunTimeoutError、RunFailedError、ProtocolError 区分等待超时、运行失败和协议异常。
• 会话与租户归属校验、鉴权、队列持久化、限流和登录恢复由业务后端负责。
• 删除 Agent 会级联删除其会话；只清理自己创建且确认不再使用的资源。

完整方法见 API.md，验证范围见 TESTING.md。