BEEOS CLOUD / DEVELOPER REFERENCE

Python 智能体运行时

Python 包根的纯标准库 BeeOSAgentRuntime 门面。

Python 包根提供 BeeOSAgentRuntime：一个把聊天消息、取消事件与运行时方法分发给你的处理函数的小型门面，并带有使用运行时 token 鉴权的辅助方法。它只依赖标准库。本模块没有对应的 TypeScript 实现。

BeeOSAgentRuntime

Python beeos_cloud_harness_sdk · 仅 Python

面向 Python 自定义智能体 harness 的门面。它向 Agent Gateway 和消息服务发送带鉴权的请求，把调用按命名空间（agent、messages、chat、files、canvas、a2a、messaging）分组，并把收到的聊天提示、取消与运行时方法分发给你注册的处理函数。

class BeeOSAgentRuntime
def __init__(agent_gateway_url: str, message_service_url: str, instance_id: str, agent_id: str, token: str | Callable[[], str], framework: str = 'custom', transport=None)

Python · 构造

参数	类型	必填	默认值	说明
`agent_gateway_url`	`str`	是		Agent Gateway 的基础 URL。
`message_service_url`	`str`	是		消息服务的基础 URL。
`instance_id`	`str`	是		实例的标识，作为 `X-BeeOS-Instance-ID` 发送。
`agent_id`	`str`	是		智能体的标识，作为 `X-BeeOS-Agent-ID` 发送。
`token`	`str \| Callable[[], str]`	是		放入 `Authorization` 头的凭据，可以是字符串，也可以是返回当前凭据的函数。已含空格的值原样发送，否则会加上 `Bearer` 前缀。
`framework`	`str`	否	`'custom'`	所接入的智能体框架名称，默认 `custom`。
`transport`		否	`None`	可选函数 `(method, url, headers, body) -> (status, raw_body, response)`，用它代替 `urllib` 发送请求，适合测试。

token

返回当前凭据：当 token 是函数时会调用它。

def token() -> str

Python · 参数

无参数。

返回值 str

register

注册收到事件时的处理函数。每次调用都会替换全部三个处理函数。

def register(*, on_chat_message=None, on_chat_cancel=None, on_runtime_method=None)

Python · 参数

参数	必填	默认值	说明
`on_chat_message`	否	`None`	收到提示时，以聊天事件字典调用。
`on_chat_cancel`	否	`None`	提示被取消时，以取消事件字典（`conversationId`、`targetMessageId`、`agentId`）调用。
`on_runtime_method`	否	`None`	每个运行时方法以 `(method, params)` 调用。未设置时，`session/prompt` 交给 `on_chat_message`，其他方法以状态 501 失败。

返回值 void

request

发送一次带鉴权的 JSON 请求，返回解码后的 JSON；空应答或 204 返回 None。错误状态码或运行时已关闭时抛出 BeeOSAgentError。

def request(base: str, method: str, path: str, json_body=None, headers=None)

Python · 参数

参数	类型	必填	默认值	说明
`base`	`str`	是		调用哪个服务：`gateway` 表示 Agent Gateway，其他值表示消息服务。
`method`	`str`	是		HTTP 方法。
`path`	`str`	是		请求路径。
`json_body`		否	`None`	与 JSON 兼容的请求体，`None` 表示没有请求体。
`headers`		否	`None`	额外的请求头。

返回值 void

dispatch_chat_message

把聊天事件交给 on_chat_message 处理函数并返回其结果；未设置处理函数时什么也不做。

def dispatch_chat_message(event: dict)

Python · 参数

参数	类型	必填	默认值	说明
`event`	`dict`	是		聊天事件，含 `conversationId`、`messageId`、`agentId`、`text`、`channelId`、`sessionKey` 与 `files`。

返回值 void

dispatch_chat_cancel

把取消事件交给 on_chat_cancel 处理函数并返回其结果；未设置处理函数时什么也不做。

def dispatch_chat_cancel(event: dict)

Python · 参数

参数	类型	必填	默认值	说明
`event`	`dict`	是		取消事件，含 `conversationId`、`targetMessageId` 与 `agentId`。

返回值 void

dispatch_runtime_method

把运行时方法调用分发给你的处理函数。方法不在 RUNTIME_METHODS 中时抛出状态 400 的 BeeOSAgentError；没有对应处理函数时抛出状态 501 的错误。

def dispatch_runtime_method(method: str, params: dict | None = None)

Python · 参数

参数	类型	必填	默认值	说明
`method`	`str`	是		HTTP 方法。
`params`	`dict \| None`	否	`None`	方法调用的参数；`None` 视为空字典。

返回值 void

dispatch_session_prompt

用 extract_session_prompt 把 session/prompt 参数转成聊天事件，并交给聊天处理函数。参数不完整时抛出状态 400 的 BeeOSAgentError。

def dispatch_session_prompt(params: dict)

Python · 参数

参数	类型	必填	默认值	说明
`params`	`dict`	是		`session/prompt` 调用的参数。

返回值 void

consume_envelope

处理一个消息信封：chat_cancel 信封转成取消事件；其他带有提示文本、消息 ID 和会话 ID 的信封转成聊天事件；其余信封被忽略并返回 None。

def consume_envelope(envelope: dict)

Python · 参数

参数	类型	必填	默认值	说明
`envelope`	`dict`	是		收到的原始消息信封，含 `type`、`conversationId`、`content` 等字段。

返回值 void

close

关闭运行时并清除处理函数。之后的请求会抛出状态 409 的 BeeOSAgentError。

def close()

Python · 参数

无参数。

返回值 void

agent.get

读取智能体记录。

def get()

Python · 参数

无参数。

返回值 void

agent.update_exposure

修改智能体的暴露设置。当前资源版本通过 If-Match 发送，因此记录期间发生变化时修改会失败。

def update_exposure(patch, resource_version)

Python · 参数

参数	类型	必填	默认值	说明
`patch`		是		要修改的字段，字典形式。
`resource_version`		是		本次修改所基于的记录资源版本。

返回值 void

agent.sync

把智能体定义同步到 Cloud。

def sync(input)

Python · 参数

参数	类型	必填	默认值	说明
`input`		是		同步请求体，字典形式。

返回值 void

messages.send

向会话发送一条消息。

def send(conversation_id, input)

Python · 参数

参数	类型	必填	默认值	说明
`conversation_id`		是		会话的标识。
`input`		是		消息体，字典形式，含 `type`、`content`、`replyTo` 等字段。

返回值 void

messages.reply

向一条收到的消息发送纯文本的 agent_reply。缺少会话 ID 或消息 ID 时抛出状态 400 的 BeeOSAgentError。

def reply(conversation_id: str, inbound_message_id: str, text: str)

Python · 参数

参数	类型	必填	说明
`conversation_id`	`str`	是	会话的标识。
`inbound_message_id`	`str`	是	被回复消息的标识。
`text`	`str`	是	回复文本。

返回值 void

messages.list

列出某个会话的消息。

def list(conversation_id)

Python · 参数

参数	类型	必填	默认值	说明
`conversation_id`		是		会话的标识。

返回值 void

messages.consume_envelope

与 BeeOSAgentRuntime.consume_envelope 相同。

def consume_envelope(envelope: dict)

Python · 参数

参数	类型	必填	默认值	说明
`envelope`	`dict`	是		收到的原始消息信封。

返回值 void

chat.patch_reply

修改 Cloud 为提示创建的回复消息，例如其正文、消息片段或状态。缺少会话 ID 或回复 ID 时抛出状态 400 的 BeeOSAgentError。

def patch_reply(conversation_id: str, reply_message_id: str, body: RuntimeMessagePatch, operation_key: str = '')

Python · 参数

参数	类型	必填	默认值	说明
`conversation_id`	`str`	是		会话的标识。
`reply_message_id`	`str`	是		回复消息的标识。
`body`	RuntimeMessagePatch	是		调用的请求体，为与 JSON 兼容的数据。
`operation_key`	`str`	否	`''`	幂等键，作为 `Idempotency-Key` 发送。为空表示不带幂等键。

返回值 void

chat.complete_reply

把回复置为 completed：以给定文本作为正文和唯一的文本片段，停止原因为 end_turn。

def complete_reply(conversation_id: str, reply_message_id: str, text: str, operation_key: str = '')

Python · 参数

参数	类型	必填	默认值	说明
`conversation_id`	`str`	是		会话的标识。
`reply_message_id`	`str`	是		回复消息的标识。
`text`	`str`	是		回复的最终文本。
`operation_key`	`str`	否	`''`	幂等键，作为 `Idempotency-Key` 发送。为空表示不带幂等键。

返回值 void

chat.fail_reply

把回复置为 failed，正文为给定文本，停止原因为 error。

def fail_reply(conversation_id: str, reply_message_id: str, text: str = '', operation_key: str = '')

Python · 参数

参数	类型	必填	默认值	说明
`conversation_id`	`str`	是		会话的标识。
`reply_message_id`	`str`	是		回复消息的标识。
`text`	`str`	否	`''`	随失败回复保存的正文，例如错误文本，默认为空。
`operation_key`	`str`	否	`''`	幂等键，作为 `Idempotency-Key` 发送。为空表示不带幂等键。

返回值 void

files.list

列出智能体的文件。

def list()

Python · 参数

无参数。

返回值 void

files.presign

为文件申请预签名上传。

def presign(input)

Python · 参数

参数	类型	必填	默认值	说明
`input`		是		预签名请求体，字典形式。

返回值 void

files.confirm

确认已上传文件完整。

def confirm(input)

Python · 参数

参数	类型	必填	默认值	说明
`input`		是		确认请求体，字典形式。

返回值 void

files.resolve

把文件 ID 解析为下载信息。

def resolve(file_id)

Python · 参数

参数	类型	必填	默认值	说明
`file_id`		是		文件的标识。

返回值 void

canvas.token

创建画布令牌。

def token(input=None)

Python · 参数

参数	类型	必填	默认值	说明
`input`		否	`None`	可选的请求体，字典形式；默认发送空请求体。

返回值 void

a2a.discover

发现可通过智能体间调用（A2A）访问的智能体。

def discover()

Python · 参数

无参数。

返回值 void

a2a.call

用 JSON-RPC 请求调用另一个智能体。

def call(agent_id, input)

Python · 参数

参数	类型	必填	默认值	说明
`agent_id`		是		被调用智能体的标识。
`input`		是		JSON-RPC 请求，字典形式。

返回值 void

a2a.get_task

读取一个智能体间任务的状态。

def get_task(task_id)

Python · 参数

参数	类型	必填	默认值	说明
`task_id`		是		任务的标识。

返回值 void

messaging.token

创建消息令牌。

def token(input=None)

Python · 参数

参数	类型	必填	默认值	说明
`input`		否	`None`	可选的请求体，字典形式；默认发送空请求体。

返回值 void

extract_chat_prompt

Python beeos_cloud_harness_sdk · 仅 Python

从消息信封中提取提示：文本、附带文件、会话 ID、会话键和消息 ID。信封中没有文本时返回 None。

def extract_chat_prompt(envelope: dict, local_agent_id: str = '') -> dict | None

Python · 参数

参数	类型	必填	默认值	说明
`envelope`	`dict`	是		收到的原始消息信封。
`local_agent_id`	`str`	否	`''`	本智能体的标识，用于构造会话键。

返回值 dict \| None

extract_session_prompt

Python beeos_cloud_harness_sdk · 仅 Python

把 session/prompt 参数转成聊天事件。缺少会话 ID、回复消息 ID、任务请求消息 ID 或运行 ID 时返回 None，以便调用方让操作失败，而不是写入未绑定的回复。

def extract_session_prompt(params: dict, local_agent_id: str = '') -> dict | None

Python · 参数

参数	类型	必填	默认值	说明
`params`	`dict`	是		`session/prompt` 调用的参数。
`local_agent_id`	`str`	否	`''`	本智能体的标识，参数中没有指定智能体时使用。

返回值 dict \| None

BeeOSAgentError

Python beeos_cloud_harness_sdk · 仅 Python

门面抛出的错误，带有 HTTP status 和机器可读的 code。

class BeeOSAgentError(RuntimeError)
def __init__(status: int, code: str, message: str)

Python · 构造

参数	类型	必填	说明
`status`	`int`	是	失败对应的 HTTP 状态码。
`code`	`str`	是	机器可读的错误码。
`message`	`str`	是	错误信息。

错误与异常

本模块定义的错误类型。每个方法的说明会写明它在什么情况下抛出哪种错误。

错误	语言	说明
BeeOSAgentError	Python	门面抛出的错误，带有 HTTP `status` 和机器可读的 `code`。

BeeOS Cloud返回文档首页