# 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 或更高版本。建议在虚拟环境中安装。

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install nexus-agent-sdk-python
export ANTHROPIC_API_KEY="替换为你的 API Key"
```

Windows PowerShell 对应命令：

```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 时使用以下配置片段：

```python
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`。这个示例只开放读取和搜索工具，逐条打印回答并检查最终结果。

```python
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` 是关键字参数。

### 持续对话

```python
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

```python
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 及兼容服务

```python
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 源码](https://github.com/nexus-research-lab/nexus-agent-sdk-python/blob/main/src/nexus_agent_sdk/options.py)。

## 消息与流式输入

| 类型 | 读取方式 |
| --- | --- |
| `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`，不要把缺失值当作零费用。

### 实时文本增量

```python
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` 可能同时出现。界面选择一种方式拼接正文，避免重复展示。

### 连续提交多个输入

```python
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 宿主中执行。

```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 服务：

```python
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 请求，并记录工具名称。真实服务还应根据工具输入限制可访问路径。

```python
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 回调源码](https://github.com/nexus-research-lab/nexus-agent-sdk-python/blob/main/src/nexus_agent_sdk/types/callbacks.py)。

## 会话恢复与本地记录

保存结果或客户端的 `session_id`，下次用 `resume` 恢复。以下配置片段中的 ID 应替换为实际已保存会话：

```python
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，可列出记录、读取消息和追加标题或标签：

```python
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` 取值。

```python
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`。以下程序用标准库超时限制任务等待，并区分结果失败与通信失败：

```python
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()` 明确释放资源，以下是函数片段：

```python
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()` 设置边界。

## 版本与进一步阅读

```bash
python -m pip show nexus-agent-sdk-python
python -c "import nexus_agent_sdk; print(nexus_agent_sdk.__version__)"
```

上线时固定已验证的 SDK 版本；使用自定义 nxs 时同时记录运行时版本。Python API 以当前安装包和公开源码为准。

- [Python SDK 仓库与安装说明](https://github.com/nexus-research-lab/nexus-agent-sdk-python)
- [Python API 参考](https://github.com/nexus-research-lab/nexus-agent-sdk-python/blob/main/docs/api-reference.md)
- [完整操作示例](https://github.com/nexus-research-lab/nexus-agent-sdk-python/tree/main/examples)
- [Python SDK 变更记录](https://github.com/nexus-research-lab/nexus-agent-sdk-python/blob/main/CHANGELOG.md)
- [Agent SDK 概览](/docs/sdk-overview)与 [Go 快速开始](/docs/sdk-quickstart)