Agent SDK
Python SDK 使用指南
安装 Python SDK，配置模型，完成多轮对话、流式输入、工具回调与会话恢复。
用 Python 构建 Agent 应用
nexus-agent-sdk-python 是 Nexus 的异步 Python 接入库，导入名为 nexus_agent_sdk。它通过 stdio stream-json 启动 nxs，由运行时完成模型调用、工具执行和上下文管理。你的 Python 应用负责提交任务、消费消息、处理权限和保存会话身份，可以独立于 Nexus 桌面应用运行。
本页包含安装、常用操作和 API 导航。标注为「配置片段」或「函数片段」的代码需放入已有程序；其余带 asyncio.run(main()) 的示例可保存为 .py 文件运行。
安装与模型凭据
需要 Python 3.11 或更高版本。建议在虚拟环境中安装。
python3 -m venv .venv
source .venv/bin/activate
python -m pip install nexus-agent-sdk-python
export ANTHROPIC_API_KEY="替换为你的 API Key"
Windows PowerShell 对应命令：
py -3 -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install nexus-agent-sdk-python
$env:ANTHROPIC_API_KEY="替换为你的 API Key"
正式平台 wheel 内置 nxs 和 rg，支持 Linux、macOS、Windows 的 x86_64 与 ARM64。模型凭据需自行配置，SDK 不提供模型额度。源码开发环境需自行提供 nxs。
运行时查找顺序是显式 cli_path、包内 nxs、开发环境的 PATH。安装包内的运行时损坏时应重新安装，SDK 不会动态下载运行时。需要指定自有 nxs 时使用以下配置片段：
from nexus_agent_sdk import NexusAgentOptions
options = NexusAgentOptions(cli_path="/path/to/nxs")
cli_path 应指向 nxs。Go Bridge 的第三方运行时适配不代表 Python SDK 可以直接切换到 Claude Code。
完成第一次查询
保存为 main.py，在包含 README.md 的项目目录运行 python main.py。这个示例只开放读取和搜索工具，逐条打印回答并检查最终结果。
import asyncio
from nexus_agent_sdk import (
AssistantMessage,
NexusAgentOptions,
ResultError,
ResultMessage,
TextBlock,
query,
)
async def main():
options = NexusAgentOptions(
cwd=".",
system_prompt="你是代码审查助手，用中文给出简明结论。",
tools=["Read", "Glob", "Grep"],
allowed_tools=["Read", "Glob", "Grep"],
max_turns=3,
)
async for message in query(
prompt="阅读 README.md，用三句话说明项目用途。", options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
elif isinstance(message, ResultMessage):
if message.is_error:
raise ResultError(message)
print("会话 ID：", message.session_id)
print("用量：", message.usage)
asyncio.run(main())
query() 返回异步消息迭代器，普通输入消费到本轮 ResultMessage 后关闭连接。它不会自动把失败结果转成异常，需要检查 is_error。收到文本表示产生了回答片段，最终是否成功以结果消息为准。
选择 query 或 NexusSDKClient
| 需求 | query() | NexusSDKClient |
|---|---|---|
| 一次性任务 | 自动管理连接 | 需进入上下文或显式连接 |
| 多轮对话 | 用 resume 或 continue_conversation 恢复 | 同一客户端保留上下文 |
| 异步迭代输入 | 支持，读取全部输入的结果 | 支持，配合 receive_messages() |
| 权限、Hooks、MCP | 通过 options 配置 | 通过 options 配置 |
| 中断和运行期控制 | 无独立控制入口 | interrupt() 等公开方法 |
两种入口的 prompt 均可使用文本、内容块列表、OutboundMessage 或异步可迭代输入。query() 的 prompt 和 options 是关键字参数。
持续对话
import asyncio
from nexus_agent_sdk import NexusAgentOptions, NexusSDKClient
async def main():
async with NexusSDKClient(NexusAgentOptions(tools=[])) as client:
await client.query("记住数字 7。")
print((await client.receive_result()).result)
await client.query("我刚才给你的数字是什么？")
print((await client.receive_result()).result)
print("恢复时保存这个 ID：", client.session_id)
asyncio.run(main())
receive_result() 消费本轮消息并返回结果，默认在失败时抛出 ResultError。需要展示过程时使用 receive_response()，它包含本轮最终结果，但需要自行检查 is_error。
同一客户端只允许一个消息消费者。不要同时调用多个接收方法，也不要在前一轮尚未完成时提交新 query。
配置 Provider 和选项
模型名称以所接服务实际支持的模型为准。以下为配置片段，创建的 options 可传给 query() 或 NexusSDKClient。
Anthropic Messages
import os
from nexus_agent_sdk import AnthropicProvider, NexusAgentOptions
options = NexusAgentOptions(
provider=AnthropicProvider(
api_key=os.environ["ANTHROPIC_API_KEY"],
base_url="https://api.anthropic.com",
),
model=os.environ["ANTHROPIC_MODEL"],
)
除 API Key 外，这段示例需要设置 ANTHROPIC_MODEL。AnthropicProvider 也支持 auth_token、version、headers 和 tool_discovery_transport。
OpenAI 及兼容服务
import os
from nexus_agent_sdk import NexusAgentOptions, OpenAIProvider
options = NexusAgentOptions(
provider=OpenAIProvider(
api_key=os.environ["OPENAI_API_KEY"],
base_url="https://api.openai.com/v1",
protocol="responses",
),
model=os.environ["OPENAI_MODEL"],
)
先设置 OPENAI_API_KEY 和 OPENAI_MODEL。protocol 可取 responses 或 chat_completions；兼容服务通过 base_url 指定，还可配置 org_id、project_id 和 headers。
Provider 字段覆盖 options.env 中对应的配置，未设置字段继续继承环境配置；非空 headers 替换继承的 headers。凭据通过运行时环境传递，不进入 CLI 参数。
NexusAgentOptions 常用字段
| 字段 | 用途与约束 |
|---|---|
cli_path、cwd、env | 运行时路径、工作目录、子进程环境 |
provider、model、fallback_model | 服务配置、主模型和回退模型 |
system_prompt、system_prompt_file | 设置系统提示词，二者互斥 |
append_system_prompt | 追加系统提示词 |
tools | 内置工具集合；[] 关闭内置工具 |
allowed_tools、disallowed_tools | 权限允许规则和拒绝规则；与工具集合分别配置 |
permission_mode | 默认 default，另有 acceptEdits、plan、auto、bypassPermissions |
max_turns、max_budget_usd | 回合与费用预算，必须为正数；回合数须为整数 |
thinking、max_thinking_tokens | 思考模式与预算，模式可取 adaptive、enabled、disabled |
resume、continue_conversation、fork_session | 恢复与分叉；前两项不能同时使用 |
persist_session | 是否持久化会话，默认 True |
include_partial_messages | 是否接收增量事件，默认 False |
output_schema | 结构化输出的 JSON Schema |
mcp_servers、agents | MCP 服务与子 Agent 定义 |
can_use_tool、hooks | 权限回调与生命周期回调 |
callback_timeout | 宿主回调超时，默认不设置 |
control_timeout、shutdown_timeout | 控制请求与关闭超时，默认分别为 60、5 秒 |
max_buffer_size、max_queue_bytes | 单条协议缓冲与消息队列限制，默认分别为 16、64 MiB |
bypassPermissions 还要求 allow_dangerously_skip_permissions=True，不作为普通接入示例的默认值。完整字段见 Options 源码。
消息与流式输入
| 类型 | 读取方式 |
|---|---|
AssistantMessage | 遍历 content，按内容块类型处理 |
TextBlock、ThinkingBlock | 文本与思考内容 |
ToolUseBlock、ToolResultBlock | 工具调用与返回内容 |
UserMessage、SystemMessage | 用户消息与系统事件 |
StreamEvent | 从 event 读取原始增量事件 |
ResultMessage | 本轮终态、结果与用量 |
TaskStartedMessage、TaskProgressMessage、TaskNotificationMessage | 后台任务事件 |
Message、UnknownBlock | 未识别消息或内容块保留原始字段 |
结果常用字段包括 is_error、subtype、result、errors、session_id、num_turns、usage、model_usage、total_cost_usd 和 structured_output。费用可能为 None，不要把缺失值当作零费用。
实时文本增量
import asyncio
from nexus_agent_sdk import NexusAgentOptions, ResultError, ResultMessage, StreamEvent, query
async def main():
options = NexusAgentOptions(tools=[], include_partial_messages=True)
async for message in query(prompt="用三句话介绍 Python。", options=options):
if isinstance(message, StreamEvent):
delta = message.event.get("delta", {})
if delta.get("type") == "text_delta":
print(delta.get("text", ""), end="", flush=True)
elif isinstance(message, ResultMessage) and message.is_error:
raise ResultError(message)
print()
asyncio.run(main())
增量事件与完整 AssistantMessage 可能同时出现。界面选择一种方式拼接正文，避免重复展示。
连续提交多个输入
import asyncio
from nexus_agent_sdk import NexusAgentOptions, NexusSDKClient, ResultError, ResultMessage
async def prompts():
yield "记住数字 7。"
yield "刚才的数字是什么？"
async def main():
async with NexusSDKClient(NexusAgentOptions(tools=[])) as client:
await client.query(prompts())
async for message in client.receive_messages():
if isinstance(message, ResultMessage):
if message.is_error:
raise ResultError(message)
print(message.result)
asyncio.run(main())
SDK 等待上一轮 result 后再发送下一条输入。这里必须消费 receive_messages() 到输入流结束，不能只读第一个结果，也不能在流结束前插入新的 query。
OutboundMessage 可携带内容块；OutboundMessageOptions 可设置单条输入的执行限制。tool_access="none"、max_output_tokens 等需要运行时协商 message_execution_policy_v1，可通过已连接客户端的 capabilities 查看。
自定义工具与 MCP
tool(name, description, input_schema) 把异步函数包装为 SdkMcpTool。schema 可用 {"a": int} 这样的类型映射，也可使用完整 JSON Schema。create_sdk_mcp_server(name=..., tools=[...]) 创建进程内 MCP 服务，工具函数在 Python 宿主中执行。
import asyncio
from nexus_agent_sdk import NexusAgentOptions, NexusSDKClient, create_sdk_mcp_server, tool
@tool("add", "计算两个整数的和", {"a": int, "b": int})
async def add(args):
return {"content": [{"type": "text", "text": str(args["a"] + args["b"])}]}
async def main():
server = create_sdk_mcp_server(name="calculator", tools=[add])
options = NexusAgentOptions(
tools=[],
mcp_servers={"calculator": server},
allowed_tools=["mcp__calculator__add"],
max_turns=3,
)
async with NexusSDKClient(options) as client:
await client.query("调用 calculator 的 add 计算 12 加 30。")
print((await client.receive_result()).result)
asyncio.run(main())
工具返回 MCP 的 content 数组。工具名称的权限规则使用 mcp__服务名__工具名。schema 校验之外，宿主仍需验证业务权限和外部资源访问范围。
外部 stdio 服务配置片段如下。将脚本路径替换为已有 MCP 服务：
import sys
from nexus_agent_sdk import NexusAgentOptions
options = NexusAgentOptions(
mcp_servers={
"business": {
"type": "stdio",
"command": sys.executable,
"args": ["/path/to/mcp_server.py"],
}
}
)
连接后可 await client.get_mcp_status() 检查状态。动态替换服务使用 await client.set_mcp_servers(servers)；这是替换配置，需传入希望保留的完整集合。动态更新超时后 SDK 会关闭会话，调用方应重建客户端。
权限回调与 Hooks
can_use_tool(name, input_data, context) 是异步权限回调，返回 PermissionResultAllow 或 PermissionResultDeny。当运行时请求宿主审批且未设置回调时，SDK 默认拒绝。
下面是配置片段，只批准到达回调的 Read、Glob 和 Grep 请求，并记录工具名称。真实服务还应根据工具输入限制可访问路径。
from nexus_agent_sdk import (
HookMatcher,
NexusAgentOptions,
PermissionResultAllow,
PermissionResultDeny,
)
async def approve_read_only(name, input_data, context):
if name in {"Read", "Glob", "Grep"}:
return PermissionResultAllow()
return PermissionResultDeny(message=f"本应用未授权工具 {name}")
async def observe_tool(input_data, tool_use_id, context):
print("准备调用：", input_data.get("tool_name"))
return {}
options = NexusAgentOptions(
tools=["Read", "Glob", "Grep"],
can_use_tool=approve_read_only,
hooks={"PreToolUse": [HookMatcher(hooks=[observe_tool], timeout=10)]},
callback_timeout=30,
)
权限回调只处理运行时转交的审批，不保证每次工具调用都会触发。PermissionResultAllow(updated_input=...) 可替换本次输入；拒绝结果可设置 interrupt=True 请求中断。
Hook 回调签名为 (input_data, tool_use_id, context)，返回协议字典。HookMatcher 可设置 matcher 和超时。权限 context 的 cancelled 与 Hook context 的 context["cancelled"] 均为取消事件；回调必须传播 asyncio.CancelledError。
其他交互入口包括 on_elicitation、on_user_dialog 和 on_oauth_token_refresh。其类型见 Python 回调源码。
会话恢复与本地记录
保存结果或客户端的 session_id，下次用 resume 恢复。以下配置片段中的 ID 应替换为实际已保存会话：
from nexus_agent_sdk import NexusAgentOptions
options = NexusAgentOptions(resume="已保存的会话 ID")
fork_options = NexusAgentOptions(resume="已保存的会话 ID", fork_session=True)
恢复需要原运行时配置目录中的 transcript 可访问。continue_conversation=True 可继续已有对话，但不能与 resume 同时设置。业务中需要精确关联时保存并使用明确的 Session ID。
SessionStore 是同步本地文件 API，可列出记录、读取消息和追加标题或标签：
from nexus_agent_sdk import SessionStore
store = SessionStore()
for session in store.list_sessions(limit=10):
print(session.session_id, session.custom_title, session.cwd)
for message in store.get_session_messages(session.session_id, limit=5):
print(message.type, message.message)
默认根目录取 NEXUS_CONFIG_DIR，未设置时为 ~/.nexus，从根下的 projects/ 读取 transcript。若运行时使用自定义目录，显式传入相同的 SessionStore(config_dir=...)。Nexus 产品托管会话应指向对应用户的 runtime 目录，不能假设顶层状态根就是会话根。
| 方法 | 用途 |
|---|---|
list_sessions(directory=..., limit=..., offset=...) | 按最近修改时间列出；directory 精确匹配工作目录 |
get_session_info(session_id) | 返回 SessionInfo，不存在时返回 None |
get_session_messages(session_id, limit=..., offset=...) | 返回用户与助手消息 |
rename_session(session_id, title) | 写入非空标题 |
tag_session(session_id, tag) | 写入标签，None 清除标签 |
结构化输出
使用 output_schema 指定 JSON Schema，从成功结果的 structured_output 取值。
import asyncio
import json
from nexus_agent_sdk import NexusAgentOptions, NexusSDKClient
async def main():
options = NexusAgentOptions(
tools=[],
max_turns=3,
output_schema={
"type": "object",
"properties": {
"name": {"type": "string"},
"language": {"type": "string"},
},
"required": ["name", "language"],
"additionalProperties": False,
},
)
async with NexusSDKClient(options) as client:
await client.query("项目名称是 Nexus，语言是 Python。提取名称和语言。")
result = await client.receive_result()
print(json.dumps(result.structured_output, ensure_ascii=False, indent=2))
asyncio.run(main())
应用应检查结构化结果是否存在，并在执行后续业务操作前验证业务约束。
中断、超时与错误处理
interrupt(reason="") 请求停止当前回合，disconnect() 关闭连接。连接生命周期优先交给 async with。以下程序用标准库超时限制任务等待，并区分结果失败与通信失败：
import asyncio
from nexus_agent_sdk import NexusSDKClient, NexusSDKError, ResultError
async def main():
try:
async with NexusSDKClient() as client:
try:
async with asyncio.timeout(60):
await client.query("用一句话介绍 Nexus。")
print((await client.receive_result()).result)
except TimeoutError:
await client.interrupt("等待超过 60 秒")
print("任务等待超时，即将关闭连接。")
except ResultError as error:
print("任务失败：", error.result.subtype, error.result.errors)
except NexusSDKError as error:
print("SDK 调用失败：", error)
asyncio.run(main())
| 异常 | 含义与处理 |
|---|---|
NexusSDKError | SDK 错误基类 |
ProcessError | 运行时启动、退出或管道失败；检查 exit_code、stderr |
ProtocolError | 运行时输出或 transcript 格式无效 |
ControlError | 控制请求被拒绝，可读取 request_id |
ResultError | receive_result() 收到失败结果；原始结果在 error.result |
BufferOverflowError | 消息积压超过队列预算，检查消费者速度 |
ValueError、TypeError | 配置或输入校验失败，修正调用参数 |
需要自行处理失败结果时，使用 await client.receive_result(raise_on_error=False) 并检查 is_error。提前退出 query() 的迭代时，用 contextlib.aclosing() 明确释放资源，以下是函数片段：
from contextlib import aclosing
from nexus_agent_sdk import query
async def read_first_message():
async with aclosing(query(prompt="你好")) as messages:
async for message in messages:
print(message)
break
运行期 API 导航
以下方法除注明同步外都需要 await。控制能力受运行时版本和协商结果影响；连接后查看 client.capabilities 并处理控制错误。
| 入口 | 用途 |
|---|---|
connect()、disconnect() | 显式打开或关闭连接 |
set_model(model)、set_permission_mode(mode) | 修改后续模型与权限策略 |
set_max_thinking_tokens(tokens) | 修改思考预算 |
update_environment(variables) | 更新运行时环境变量 |
get_context_usage()、get_settings() | 读取上下文用量与有效设置 |
apply_flag_settings(settings) | 应用运行期设置 |
set_next_turn_context(context)、clear_next_turn_context() | 同步方法，设置或清除下一轮宿主上下文 |
rewind_files(user_message_id, dry_run=True) | 预览文件检查点回滚；实际回滚需关闭 dry run |
remove_messages(message_uuids) | 删除对话消息，不回滚外部副作用 |
stop_task(task_id)、send_task_message(...) | 停止后台任务或发消息 |
get_mcp_status()、set_mcp_servers(servers) | 查询或替换 MCP 配置 |
authenticate_mcp(...)、submit_mcp_oauth_callback(...) | 发起认证并提交 OAuth 回调 |
clear_mcp_auth(...)、reconnect_mcp(...)、set_mcp_enabled(...) | 清理凭据、重连、启停服务 |
get_server_info()、supported_models()、supported_commands()、supported_agents()、account_info() | 同步读取初始化信息 |
agents 可注册 AgentDefinition，其中 description 和 prompt 必填，还可指定工具、模型与回合限制。高级 control_subagent() 需要处于活动 MCP 工具调用中，通过 get_tool_context().tool_use_id 获取调用 ID；等待此操作或 run_auto_dream() 时用 asyncio.timeout() 设置边界。
版本与进一步阅读
python -m pip show nexus-agent-sdk-python
python -c "import nexus_agent_sdk; print(nexus_agent_sdk.__version__)"
上线时固定已验证的 SDK 版本；使用自定义 nxs 时同时记录运行时版本。Python API 以当前安装包和公开源码为准。