comate-agent-sdk
一个用于构建 Agent(工具调用 + for-loop)的 Python SDK。它强调“行动空间完整 + 显式退出 + 上下文工程”,同时尽量保持 API 简洁、可维护、可扩展。
你能用它做什么
- 把任意
async def函数变成可被 LLM 调用的工具:签名 → JSON Schema(支持 Pydantic 参数模型) - FastAPI 风格依赖注入:
Depends(...)(适合注入 DB/客户端/配置/上下文) - 上下文工程:自动压缩(compaction)、工具输出的 ephemeral 保留、超大内容 offload 到文件系统
- 会话能力:
ChatSession持久化 / 恢复 / 分叉(resume/fork) - 可扩展能力包:
- Subagent:通过
.agent/agents/*.md声明,自动注入Agent工具 - Skill:通过
.agent/skills/*/SKILL.md声明,自动注入Skill工具
- Subagent:通过
- Token 统计与可选计费(本地缓存定价数据)
Token 计费字段
TokenCost.get_model_pricing() 返回的 ModelPricing 使用 USD / 1M tokens 作为公开价格单位:
input_usd_per_1m_tokensoutput_usd_per_1m_tokenscache_read_usd_per_1m_tokenscache_creation_usd_per_1m_tokens
旧的 pydantic view 字段 input_cost_per_token、output_cost_per_token、
cache_read_input_token_cost 和 cache_creation_input_token_cost 已有意移除。
LiteLLM 上游定价数据仍可能使用 per-token 原始字段,SDK 会在内部 normalize 后再投影成 ModelPricing。
安装
SDK 作为依赖接入(推荐)
uv add comate-agent-sdk
CLI 快速运行(推荐)
uvx comate-cli
CLI 长期安装
uv tool install comate-cli
comate
一键安装脚本
bash scripts/install_comate.sh
在本仓库开发
uv sync
uv run pytest -q
uv run python examples/comate.py
准备环境(以 OpenAI 为例)
SDK 会在 import 时自动加载 .env(依赖 python-dotenv),所以你可以在项目根目录放一个 .env:
OPENAI_API_KEY=...
如果你要使用内置 WebFetch(系统工具),它会调用 llm_levels["LOW"],推荐你显式配置三档模型,避免默认落到 Anthropic:
COMATE_AGENT_SDK_LLM_LOW="openai:gpt-4o-mini"
COMATE_AGENT_SDK_LLM_MID="openai:gpt-4o"
COMATE_AGENT_SDK_LLM_HIGH="openai:gpt-4o"
COMATE_AGENT_SDK_LLM_LOW_BASE_URL="http://192.168.100.1:4141/v1"
COMATE_AGENT_SDK_LLM_MID_BASE_URL="http://192.168.100.1:4142/v1"
COMATE_AGENT_SDK_LLM_HIGH_BASE_URL="http://192.168.100.1:4143/v1"
facade 层也可以直接在 AgentOptions 中配置。只传 model 时,SDK 会把
LOW、MID、HIGH 三档都兜底到同一个模型;如果需要按档位区分,再用
llm_levels 覆盖具体档位:
from comate_agent_sdk.facade import Agent, AgentOptions
agent = Agent(
options=AgentOptions(
tools="comate",
model="openai:gpt-4o",
llm_levels={
"LOW": "openai:gpt-4o-mini",
"HIGH": "openai:gpt-4o",
},
),
)
上例中主 Agent 使用 MID=openai:gpt-4o;subagent 如果动态传入
level="HIGH",会使用 llm_levels["HIGH"],不会落回默认 Anthropic。
配置文件:settings.json 和 AGENTS.md
SDK 支持通过配置文件管理 LLM 配置和 Agent 指令,分为 user 级(全局)和 project 级(项目)两个层级:
| 配置文件 | User 级(全局) | Project 级 | 优先级 |
|---|---|---|---|
settings.json |
~/.agent/settings.json |
{项目根}/.agent/settings.json |
project 字段级覆盖 user |
AGENTS.md |
~/.agent/AGENTS.md |
{项目根}/AGENTS.md 或 {项目根}/.agent/AGENTS.md |
project 完全替代 user |
settings.json 配置模板
{
"llm_levels": {
"LOW": "openai:gpt-4o-mini",
"MID": "openai:gpt-4o",
"HIGH": "anthropic:claude-opus-4-5"
},
"llm_levels_base_url": {
"LOW": "http://192.168.100.1:4141/v1",
"MID": null,
"HIGH": "https://api.anthropic.com"
},
"env": {
"COMATE_AGENT_SDK_MEMORY_BACKGROUND_ENABLED": "false"
}
}
说明:
llm_levels:三档模型配置,格式为"provider:model"(provider可选:openai、anthropic、google)llm_levels_base_url:可选,为每档模型指定 base_url(null表示使用默认)env:可选,string -> string字典。会覆盖当前系统环境变量,并参与 SDK 内部 env 解析- 优先级(从高到低):代码参数
llm_levels=> project settings.json > user settings.json > 环境变量 > 默认值
settings.json.env 与 facade 的优先级
对 SDK 的 env 解析(例如 memory 后台总开关):
os.environ < settings.json.env(project/local 覆盖 user) < AgentOptions.env
如果你在 facade 中显式传 memory_background_enabled,它的优先级还会高于上述 env 体系。
记忆后台能力总开关
relevant memory side query 和 background extraction 共用一个总开关:
- env 名称:
COMATE_AGENT_SDK_MEMORY_BACKGROUND_ENABLED - 默认值:开启
最小示例:
from comate_agent_sdk.facade import AgentClient, AgentOptions
client = AgentClient(
options=AgentOptions(
model=...,
memory_background_enabled=False,
),
)
字段级覆盖示例
假设你有:
~/.agent/settings.json(user 级):
{
"llm_levels": {
"LOW": "openai:gpt-4o-mini",
"MID": "openai:gpt-4o",
"HIGH": "openai:gpt-4o"
},
"llm_levels_base_url": {
"LOW": "http://192.168.100.1:4141/v1"
}
}
{项目根}/.agent/settings.json(project 级):
{
"llm_levels": {
"HIGH": "anthropic:claude-opus-4-5"
}
}
最终生效:
llm_levels:project 的llm_levels完全覆盖 user(只有HIGH生效,LOW和MID消失)llm_levels_base_url:project 未定义,回退到 user 的配置
AGENTS.md 加载规则
AGENTS.md 是用于写入 Agent 背景指令的 Markdown 文件,会被自动注入到 memory(类似 system prompt,但不计入 prompt token 限制)。
搜索路径(按优先级):
-
Project 级(任一存在即生效):
{项目根}/AGENTS.md{项目根}/.agent/AGENTS.md
-
User 级(仅当 project 级不存在时 fallback):
~/.agent/AGENTS.md
重要规则:
- 当 project 级存在任何
AGENTS.md时,user 级会被完全忽略(不会合并) - 只有在 project 级完全不存在时,才会 fallback 到 user 级
- 如果用户代码手动指定了
memory=...,则不会自动加载任何AGENTS.md
AGENTS.md 示例
~/.agent/AGENTS.md(user 级,全局指令):
# 全局 Agent 规则
- 所有代码必须使用 f-string
- 必须使用 logging 模块,禁止 print
{项目根}/.agent/AGENTS.md(project 级,项目特定指令):
# 本项目 Agent 规则
这是一个 Django 项目,请遵循:
- 使用 Django ORM 查询数据库
- 在 views.py 中编写视图函数
- 测试文件放在 tests/ 目录
控制配置加载:setting_sources
你可以在代码中显式控制加载哪些配置:
from comate_agent_sdk.agent import Agent, AgentConfig
# 默认:加载 user 和 project 两个层级
agent = Agent(llm=..., config=AgentConfig(setting_sources=("user", "project")))
# 只加载 project 级配置
agent = Agent(llm=..., config=AgentConfig(setting_sources=("project",)))
# 只加载 user 级配置
agent = Agent(llm=..., config=AgentConfig(setting_sources=("user",)))
# 完全不加载配置文件(向后兼容模式)
agent = Agent(llm=..., config=AgentConfig(setting_sources=None))
注意:setting_sources 同时控制 settings.json 和 AGENTS.md 的加载范围。
MCP:接入外部工具(stdio/sse/http)与本地 SDK Server
SDK 支持通过 MCP(Model Context Protocol)接入外部工具生态。MCP tools 会被映射为普通 Tool,并遵循统一命名规则:
- 所有 MCP tools 都以
mcp__开头 - 命名格式:
mcp__{server_alias}__{tool_name}server_alias来自AgentConfig(mcp_servers={...})的 key,或.mcp.json里的 keytool_name来自 MCP server 返回的原始 tool name- 两段都会被规范化为仅含
[A-Za-z0-9_]的形式(如agoal-reports→agoal_reports,连续_会被压缩)
- 在
tools=[...]白名单和disallowed_tools里书写形状完整的 MCP 工具名时,写原始 alias(mcp__agoal-reports__search)或规范化形式(mcp__agoal_reports__search)均可,SDK 会在各自入口统一规范化 - 白名单条目规范化后仍匹配不到已配置 server 时会输出 warning(附当前已配置 server 的规范化 alias 列表)并跳过该条目;
disallowed_tools不做 server 存在性检查 __是工具名格式的分隔符;原始 alias 自身含__时存在语法歧义,白名单和disallowed_tools必须使用规范化 alias(如a__b写成a_b)
说明:SDK 会在第一次调用 LLM 前懒加载 MCP tools;如果你使用
ChatSession.resume()恢复会话,SDK 会在下一次调用 LLM 前自动刷新 MCP tools。
1) 配置 MCP server(stdio / sse / http)
自动发现以下三个配置文件,优先级从低到高:
- User 级:
~/.agent/.mcp.json - Project 级:
{项目根}/.agent/.mcp.json - Local 级:
{项目根}/.agent/.mcp.local.json
磁盘文件唯一合法的顶层字段是 mcpServers:
{
"mcpServers": {
"fs": { "type": "stdio", "command": "python", "args": ["-m", "my_fs_mcp_server"] }
}
}
stdio 示例(本机启动子进程)
{
"mcpServers": {
"calc": {
"type": "stdio",
"command": "python",
"args": ["-m", "my_calc_mcp_server"],
"env": { "LOG_LEVEL": "INFO" }
}
}
}
type对 stdio 可省略(缺省即按 stdio 处理)。
SSE 示例(远程/本地 SSE 端点)
{
"mcpServers": {
"search": {
"type": "sse",
"url": "http://127.0.0.1:8000/sse",
"headers": { "Authorization": "Bearer YOUR_TOKEN" }
}
}
}
HTTP 示例(Streamable HTTP 端点)
{
"mcpServers": {
"internal": {
"type": "http",
"url": "http://127.0.0.1:8000/mcp",
"headers": { "X-API-Key": "YOUR_KEY" }
}
}
}
Facade 来源组合与严格隔离
Facade 默认按 plugin < user < project < local < explicit 组合来源;同名 alias
按整项覆盖,不做字段级深合并。AgentOptions.mcp_servers 只接受 dict 或 None,
不接受配置文件 Path。
| Facade 配置 | 结果 |
|---|---|
未传 mcp_servers |
plugin + user + project + local |
mcp_servers={...} |
以上来源 + 显式 overlay |
mcp_servers={} |
空 overlay;较低来源仍生效 |
strict_mcp_config=True |
只使用显式 dict;忽略 plugin 与磁盘 MCP |
strict_mcp_config=True, mcp_servers={} |
完全不启用 MCP server |
from comate_agent_sdk.facade import Agent, AgentOptions
from comate_agent_sdk.llm import ChatOpenAI
agent = Agent(
options=AgentOptions(
model=ChatOpenAI(model="gpt-4o-mini"),
mcp_servers={
"internal": {"type": "http", "url": "http://127.0.0.1:8000/mcp"},
},
tools="comate",
),
)
要遮蔽较低来源的同名 server,在更高来源写最小 tombstone:
{
"mcpServers": {
"internal": {"enabled": false}
}
}
配置文件路径仅由直接 AgentConfig 的 legacy 模式支持:
from comate_agent_sdk.agent import Agent, AgentConfig
agent = Agent(llm=..., config=AgentConfig(mcp_servers="/abs/path/to/.mcp.json", tools=[...]))
注意:
.mcp.json不支持type="sdk"(因为instance无法序列化),sdk只能代码注入。
2) 创建本地 MCP server(SDK in-process,FastMCP)
如果你希望把一组工具“像 MCP server 一样”以内嵌方式提供给 Agent,可以用 create_sdk_mcp_server()。
要点:
- 使用
@mcp_tool(name=..., description=...)声明工具 - 输入 schema 来自函数签名的类型注解(推荐显式参数,不要用
args: dict) - 注册到 Agent 时,用
mcp_serversdict 的 key 作为server_alias(决定最终 tool name 前缀)
import asyncio
import logging
from comate_agent_sdk.facade import Agent, AgentOptions
from comate_agent_sdk.llm import ChatOpenAI
from comate_agent_sdk.mcp import create_sdk_mcp_server, mcp_tool
logging.basicConfig(level=logging.INFO)
@mcp_tool(name="add", description="Add two numbers")
async def add(a: float, b: float) -> str:
return f"Sum: {a + b}"
@mcp_tool(name="multiply", description="Multiply two numbers")
async def multiply(a: float, b: float) -> str:
return f"Product: {a * b}"
calculator = create_sdk_mcp_server(
name="calculator",
version="2.0.0",
tools=[add, multiply],
)
async def main() -> None:
agent = Agent(
options=AgentOptions(
model=ChatOpenAI(model="gpt-4o-mini"),
mcp_servers={"calc": calculator}, # alias = "calc"
tools="comate",
),
)
events = [
event
async for event in agent.run_message("请用工具计算 12.5 + 3.5,然后再乘以 2。")
]
logging.info(events)
await agent.shutdown()
if __name__ == "__main__":
asyncio.run(main())
运行:
uv run python your_script.py
快速上手:Claude Code 风格(系统工具 + 显式 done + Session)
下面这个最小示例具备:
- Claude Code 风格系统工具(
Bash/Read/Write/Edit/Grep/Glob/TodoWrite/WebFetch) - 显式
done工具(防止“无工具调用就提前结束”) ChatSession(会话持久化到~/.agent/sessions/<cwd_key>/<session_id>/)
系统工具依赖(按工具)
说明:Python 包依赖会由
uv sync/uv add comate-agent-sdk自动安装;外部命令依赖需要你在系统层安装。
| 工具 | 依赖 | 缺失时行为 | 备注 |
|---|---|---|---|
Bash |
系统 shell(Linux/macOS 默认具备) | 无法执行命令 | 具体依赖取决于你在命令里调用的程序(如 git / uv / python 等) |
Read |
无 | - | - |
Write |
无 | - | - |
Edit |
无 | - | - |
Glob |
无 | - | - |
Grep |
可选:rg(ripgrep) |
若未安装 rg,会自动回退到 Python 实现(较慢,部分高级能力可能缺失) |
推荐安装 ripgrep:Ubuntu/Debian sudo apt-get install ripgrep;macOS brew install ripgrep;Windows winget install BurntSushi.ripgrep.MSVC |
TodoWrite |
无 | - | 会在 session 目录下写入/清理 todos.json |
WebFetch |
Python 包:curl-cffi、markdownify;以及 llm_levels["LOW"] |
缺包则报错;未配置 LOW 模型则无法按预期调用低档模型 | 需要联网访问目标 URL;会调用低档模型做摘要/抽取 |
import asyncio
import logging
from comate_agent_sdk.agent import Agent, AgentConfig, SessionInitEvent, StopEvent, TextEvent, ToolCallEvent, ToolResultEvent
from comate_agent_sdk.llm import ChatOpenAI
from comate_agent_sdk.tools import get_default_registry
logging.basicConfig(level=logging.INFO)
async def main() -> None:
llm_levels = {
"LOW": ChatOpenAI(model="gpt-4o-mini"),
"MID": ChatOpenAI(model="gpt-4o"),
"HIGH": ChatOpenAI(model="gpt-4o"),
}
agent = Agent(
llm=llm_levels["MID"],
config=AgentConfig(
llm_levels=llm_levels,
tools=get_default_registry().all(),
include_cost=False,
),
)
session = agent.chat()
prompt = (
"请在当前项目根目录里:\n"
"1) 找到 pyproject.toml\n"
"2) 读取并告诉我 [project] 的 name\n"
"3) 完成后给出结论\n"
)
async for event in session.query_stream(prompt):
if isinstance(event, SessionInitEvent):
logging.info(f"session_id={event.session_id}")
elif isinstance(event, ToolCallEvent):
logging.info(f"→ {event.tool}: {event.args}")
elif isinstance(event, ToolResultEvent):
logging.info(f"← {event.tool}: is_error={event.is_error}")
elif isinstance(event, TextEvent):
logging.info(event.content)
elif isinstance(event, StopEvent):
logging.info(f"done reason={event.reason}")
if __name__ == "__main__":
asyncio.run(main())
运行方式:
uv run python your_script.py
核心概念与 API
Agent / query / AgentClient(公开 facade)
query(...):最轻量的一次性事件流入口。Agent.run_message(...):复用同一个 Agent 会话。AgentClient:event-pump 客户端;query()只发送消息,事件从receive_response()/receive_messages()读取。
AgentClient 多轮复用
import asyncio
import logging
from comate_agent_sdk.facade import AgentClient, AgentOptions
async def main() -> None:
client = AgentClient(options=AgentOptions(model=..., cwd="."))
await client.connect()
try:
await client.query("Analyze this repository")
first = [event async for event in client.receive_response()]
logging.info(first)
await client.query("Continue using the same history")
second = [event async for event in client.receive_response()]
logging.info(second)
finally:
await client.disconnect()
asyncio.run(main())
同一 connection 内的多次 query() 共享一个 ChatSession;跨进程恢复可在构造
AgentClient(session_id=...) 时指定会话 ID。
Tool:@tool(...)
- 直接装饰
async def,自动:- 解析函数签名并生成 JSON Schema
- 支持 Pydantic 参数模型(适合复杂输入)
- 支持依赖注入(
typing.Annotated[..., Depends(...)]或默认值Depends(...))
ephemeral=<N>:仅保留最近 N 次该工具输出,旧输出会被标记 destroyed,并(可选)offload 到文件系统
Depends(依赖注入)
from typing import Annotated
from comate_agent_sdk import Depends, tool
def get_db() -> "Database":
return Database()
@tool("查询用户")
async def get_user(user_id: int, db: Annotated["Database", Depends(get_db)]) -> str:
return await db.find(user_id)
Session:持久化 / 恢复 / 分叉
ChatSession 会把对话增量写入:
~/.agent/sessions/<cwd_key>/<session_id>/context.jsonl- 以及(若发生 offload)
~/.agent/sessions/<cwd_key>/<session_id>/offload/
恢复(resume)
session = agent.chat(session_id="<已有 session_id>")
分叉(fork)
forked = agent.chat(fork_session="<已有 session_id>")
获取统计和清空历史
# 获取 token 使用统计
usage = await session.get_usage()
print(f"总 tokens: {usage.total_tokens}")
print(f"总成本: ${usage.total_cost:.4f}")
# 按模型查看
for model, stats in usage.by_model.items():
print(f"{model}: {stats.total_tokens} tokens")
# 清空会话历史(包括 token 统计和持久化)
session.clear_history()
注意:
get_usage()即使在 session 关闭后也可调用(用于获取最终统计)clear_history()会清空内存、token 统计,并向 JSONL 写入重置事件- 如需保留历史,请先使用
fork_session()创建副本
Context:Compaction / Offload / Ephemeral
1) 自动压缩(Compaction)
from comate_agent_sdk.agent import AgentConfig, CompactionConfig
agent = Agent(
llm=ChatOpenAI(model="gpt-4o"),
config=AgentConfig(
tools=[...],
compaction=CompactionConfig(threshold_ratio=0.80),
emit_compaction_meta_events=False, # 调试事件开关(默认关闭)
),
)
当前压缩行为(直接替换旧策略):
- 工具历史按“工具块”处理:仅保留最近 5 块,更早块整块删除
- 保留块内字段阈值截断:
tool_call.arguments > 500 tokens才截断tool_result.content > 600 tokens才截断- 截断保留前 200 tokens,并追加
[TRUNCATED original~N tokens]
user/assistant至少保留最近 12 轮(is_meta=True不计轮次)- 选择性压缩后始终执行 summary;任一步失败原子回滚
- summary 失败会自动短重试;连续失败进入短冷却,避免高频重复失败
2) Offload(卸载到文件系统)
默认开启(offload_enabled=True),并写入:
~/.agent/sessions/<cwd_key>/<session_id>/offload/
常用配置:
from comate_agent_sdk.agent import AgentConfig
agent = Agent(
llm=ChatOpenAI(model="gpt-4o"),
config=AgentConfig(
cwd=Path.cwd(),
tools=[...],
offload_enabled=True,
offload_token_threshold=2000,
),
)
3) Ephemeral(工具输出只保留最近 N 条)
from comate_agent_sdk import tool
@tool("读取大文件(只保留最近 2 次输出)", ephemeral=2)
async def read_big(path: str) -> str:
return "..."
4) 相关文档
Subagent:.agent/agents/*.md + Agent
在项目根目录创建 .agent/agents/,每个文件一个 subagent,例如 .agent/agents/researcher.md:
---
name: researcher
description: 做信息收集与整理
tools:
- WebFetch
- Grep
- Read
level: LOW
max_iterations: 30
timeout: 60
---
你是一个研究员,输出需要结构化、可复用。
Subagent 模型配置
Subagent 支持两种方式指定使用的 LLM 模型:
1. 使用档位 (level)
推荐使用档位来控制模型性能和成本:
---
name: quick-helper
description: 快速助手
level: LOW # 使用低档位(快速、便宜)
---
支持的档位:
LOW: 快速模型(如 haiku)MID: 标准模型(如 sonnet)HIGH: 高性能模型(如 opus)
这些档位会从 llm_levels 取实际模型。facade 中只传 AgentOptions.model 时,
LOW/MID/HIGH 会全部兜底到同一个模型;如果需要不同档位,请配置
AgentOptions.llm_levels。
2. 使用别名 (model)
也可以使用别名直接指定:
---
name: expert
description: 专家
model: opus # 使用opus模型
---
支持的别名:
sonnet: 映射到MID档位opus: 映射到HIGH档位haiku: 映射到LOW档位inherit: 继承父agent(等同于不指定)
3. 默认行为
如果不指定 model 或 level,subagent 将继承父 agent 的模型。
注意: 不支持直接指定完整的模型名称(如 model: gpt-4o),仅支持上述别名。
启动后(只要项目里发现了 subagents),主 Agent 会自动注入 Agent 工具,模型即可调用:
Agent(subagent_type="researcher", prompt="...", description="...")
自动发现与 agents 语义(重要)
Subagent 支持自动发现(从文件系统加载)与代码显式传入两种方式,并且可以混用。
自动发现路径与优先级
SDK 会从以下路径发现 subagent 定义(.md 文件):
- Project 级(优先):
{project_root}/.agent/agents/*.md - User 级(fallback):
~/.agent/agents/*.md
优先级规则:
- 当 project 级存在任意
.md文件时,完全忽略 user 级(不会合并)。 project_root未显式传入时,默认使用当前工作目录(cwd)。
AgentConfig(agents=...) 的约定
agents 参数用于控制是否启用/合并 subagent(注意 None 与 [] 的语义不同):
| 传参 | 行为 | 典型用途 |
|---|---|---|
agents=None(默认) |
允许自动发现;若发现到 subagent,会注入系统 Task 工具 |
纯自动发现 |
agents=[] |
显式禁用自动发现;不会注入系统 Task 工具 |
测试隔离 / 完全不启用 subagent |
agents=[...](非空) |
自动发现 + 代码传入 合并(同名以代码传入为准) | 混合模式(推荐) |
Task 是保留名:避免静默覆盖
当启用了 subagent(最终 agent.agents 非空)时,SDK 会注入系统级 Task 工具作为 subagent 调度入口,因此:
- 若你在
tools=[...]中手动提供了同名工具Tool(name="Task"),SDK 会直接抛出ValueError(避免静默替换导致误用)。
解决方式(三选一):
- 将你的工具改名(不要叫
Agent) - 显式禁用 subagent:
Agent(..., config=AgentConfig(agents=[])) - 移除/调整自动发现的 subagent定义(例如删除/修改
.agent/agents/*.md)
Skill:.agent/skills/*/SKILL.md + Skill
在项目根目录创建 .agent/skills/<skill_name>/SKILL.md,例如 .agent/skills/release/SKILL.md:
---
name: release
description: 发布流程规范
---
发布步骤:
1) ...
2) ...
项目路径:{baseDir}
SDK 会自动发现 Skills 并注入:
Skillmeta-tool(用于加载某个 skill 的完整指令)- skill 策略提示(写入 Context header)
模型可调用:
Skill(skill_name="release")
可观测性:Langfuse 集成
SDK 支持通过 Langfuse 进行 LLM 调用的可观测性监控(tracing、usage、latency 等)。
启用方式
只需设置以下三个环境变量(必须全部设置才会启用):
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_BASE_URL=https://cloud.langfuse.com # 或你的自托管地址
设置后,SDK 会自动启用 Langfuse 追踪:
- OpenAI:使用 Langfuse 包装的 OpenAI 客户端
- Anthropic:使用 OpenTelemetry instrumentation
Anthropic 额外依赖
如果使用 Anthropic 模型并希望启用 Langfuse 追踪,需要额外安装:
pip install opentelemetry-instrumentation-anthropic
# 或
uv add opentelemetry-instrumentation-anthropic
注意:如果未安装此包但设置了环境变量,SDK 会输出警告日志但不会影响正常运行。
不启用时的行为
如果未设置(或未完整设置)上述三个环境变量,SDK 会使用原生的 LLM 客户端,不会有任何额外开销。
Token 统计与可选计费
只统计 tokens(默认)
summary = await agent.get_usage()
计算成本(需要拉取定价并缓存)
- 代码层:
Agent(config=AgentConfig(include_cost=True, ...)) - 或环境变量:
comate_agent_sdk_CALCULATE_COST=true
定价数据会缓存到 XDG_CACHE_HOME(默认 ~/.cache/comate_agent_sdk/token_cost/)。
示例代码
仓库内已有更完整示例:
comate_agent_sdk/examples/claude_code.py:Claude Code 风格(沙盒文件系统 + 依赖注入)comate_agent_sdk/examples/chat_session_repl.py:Session REPLcomate_agent_sdk/examples/chat_session_repl_fork.py:Session 分叉 REPLcomate_agent_sdk/examples/subagent_example.py:Subagent 示例comate_agent_sdk/examples/dependency_injection.py:依赖注入示例
运行(示例):
uv run python comate_agent_sdk/examples/claude_code.py
许可证
MIT
Release files for comate-agent-sdk 0.12.9
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| comate_agent_sdk-0.12.9.tar.gz | 3.1 MB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| comate_agent_sdk-0.12.9-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 6.9 MB
Release files / comate_agent_sdk-0.12.9.tar.gz
| Download URL | comate_agent_sdk-0.12.9.tar.gz |
|---|---|
| Size | 3.1 MB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
ed463836946c392d7d46f9df8a0a422bee93c5d2756b3bfb83381a0126133dec
|
|
BLAKE2b-256 checksum How to use checksums |
90bd111cc45b2b6f3f6523da408e43c6936dbb93b6f080316f9b060f7722f125
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.13.9
|
Release files / comate_agent_sdk-0.12.9-py3-none-any.whl
| Download URL | comate_agent_sdk-0.12.9-py3-none-any.whl |
|---|---|
| Size | 3.8 MB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
a3770ee30dfac5a2d77bbf5cc62c6db5712ad1b12375051fda822a46d85afdc4
|
|
BLAKE2b-256 checksum How to use checksums |
4630fe0dd0da63b4392b5e1061f2d680efc8097e471de19c65cba0c2f8713e9c
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.2.0 CPython/3.13.9
|