BEEOS CLOUD / DEVELOPER REFERENCE

Python agent runtime

The stdlib-only BeeOSAgentRuntime facade of the Python package root.

The Python package root exposes BeeOSAgentRuntime, a small facade that dispatches chat messages, cancellations and runtime methods to your handlers and offers lease-token-authenticated helpers. It uses only the standard library. This module has no TypeScript counterpart.

BeeOSAgentRuntime

Python beeos_cloud_harness_sdk · Python only

Facade for a custom agent harness in Python. It sends authenticated requests to Agent Gateway and the message service, groups the calls into namespaces (agent, messages, chat, files, canvas, a2a, messaging), and routes inbound chat prompts, cancellations and runtime methods to the handlers you register.

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 · Constructor

Parameter	Type	Required	Default	Description
`agent_gateway_url`	`str`	yes		Base URL of Agent Gateway.
`message_service_url`	`str`	yes		Base URL of the message service.
`instance_id`	`str`	yes		Identifier of the instance; sent as `X-BeeOS-Instance-ID`.
`agent_id`	`str`	yes		Identifier of the agent; sent as `X-BeeOS-Agent-ID`.
`token`	`str \| Callable[[], str]`	yes		Credential sent in the `Authorization` header, either a string or a function that returns the current one. A value that already contains a space is sent as it is; otherwise `Bearer` is added.
`framework`	`str`	no	`'custom'`	Name of the agent framework that is being connected. Defaults to `custom`.
`transport`		no	`None`	Optional function `(method, url, headers, body) -> (status, raw_body, response)` that sends the request instead of `urllib`; useful in tests.

token

Returns the current credential: calls the function when token is a function.

def token() -> str

Python · Parameters

No parameters.

Returns str

register

Registers the handlers for inbound events. Each call replaces all three handlers.

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

Python · Parameters

Parameter	Required	Default	Description
`on_chat_message`	no	`None`	Called with a chat event dict when a prompt arrives.
`on_chat_cancel`	no	`None`	Called with a cancel event dict (`conversationId`, `targetMessageId`, `agentId`) when a prompt is cancelled.
`on_runtime_method`	no	`None`	Called with `(method, params)` for every runtime method. When it is not set, `session/prompt` goes to `on_chat_message` and other methods fail with status 501.

Returns void

request

Sends one authenticated JSON request and returns the decoded JSON, or None for an empty or 204 answer. Raises BeeOSAgentError for an error status or when the runtime is closed.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`base`	`str`	yes		Which service to call: `gateway` for Agent Gateway, anything else for the message service.
`method`	`str`	yes		HTTP method.
`path`	`str`	yes		Request path.
`json_body`		no	`None`	JSON-compatible request body, or `None` for no body.
`headers`		no	`None`	Additional request headers.

Returns void

dispatch_chat_message

Passes a chat event to the on_chat_message handler and returns its result; does nothing when no handler is set.

def dispatch_chat_message(event: dict)

Python · Parameters

Parameter	Type	Required	Default	Description
`event`	`dict`	yes		Chat event with `conversationId`, `messageId`, `agentId`, `text`, `channelId`, `sessionKey` and `files`.

Returns void

dispatch_chat_cancel

Passes a cancel event to the on_chat_cancel handler and returns its result; does nothing when no handler is set.

def dispatch_chat_cancel(event: dict)

Python · Parameters

Parameter	Type	Required	Default	Description
`event`	`dict`	yes		Cancel event with `conversationId`, `targetMessageId` and `agentId`.

Returns void

dispatch_runtime_method

Routes a runtime method call to your handler. Raises BeeOSAgentError with status 400 for a method that is not in RUNTIME_METHODS, and with status 501 when no handler exists for it.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`method`	`str`	yes		HTTP method.
`params`	`dict \| None`	no	`None`	Parameters of the method call; `None` is treated as an empty dict.

Returns void

dispatch_session_prompt

Converts the session/prompt parameters into a chat event with extract_session_prompt and passes it to the chat handler. Raises BeeOSAgentError with status 400 when the parameters are incomplete.

def dispatch_session_prompt(params: dict)

Python · Parameters

Parameter	Type	Required	Default	Description
`params`	`dict`	yes		Parameters of the `session/prompt` call.

Returns void

consume_envelope

Handles a message envelope: a chat_cancel envelope becomes a cancel event; any other envelope with a prompt, a message id and a conversation id becomes a chat event. Anything else is ignored and None is returned.

def consume_envelope(envelope: dict)

Python · Parameters

Parameter	Type	Required	Default	Description
`envelope`	`dict`	yes		Message envelope as received, with `type`, `conversationId`, `content` and related fields.

Returns void

close

Closes the runtime and clears the handlers. Later requests raise BeeOSAgentError with status 409.

def close()

Python · Parameters

No parameters.

Returns void

agent.get

Reads the agent record.

def get()

Python · Parameters

No parameters.

Returns void

agent.update_exposure

Changes the exposure settings of the agent. The current resource version is sent in If-Match, so the change fails when the record changed meanwhile.

def update_exposure(patch, resource_version)

Python · Parameters

Parameter	Type	Required	Default	Description
`patch`		yes		The fields to change, as a dict.
`resource_version`		yes		Resource version of the record that the change is based on.

Returns void

agent.sync

Synchronises the agent definition with Cloud.

def sync(input)

Python · Parameters

Parameter	Type	Required	Default	Description
`input`		yes		Sync request body as a dict.

Returns void

messages.send

Posts a message to a conversation.

def send(conversation_id, input)

Python · Parameters

Parameter	Type	Required	Default	Description
`conversation_id`		yes		Identifier of the conversation.
`input`		yes		Message body as a dict, with fields such as `type`, `content` and `replyTo`.

Returns void

messages.reply

Posts a plain-text agent_reply to an inbound message. Raises BeeOSAgentError with status 400 when a conversation id or message id is missing.

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

Python · Parameters

Parameter	Type	Required	Description
`conversation_id`	`str`	yes	Identifier of the conversation.
`inbound_message_id`	`str`	yes	Identifier of the message that is answered.
`text`	`str`	yes	Text of the reply.

Returns void

messages.list

Lists the messages of a conversation.

def list(conversation_id)

Python · Parameters

Parameter	Type	Required	Default	Description
`conversation_id`		yes		Identifier of the conversation.

Returns void

messages.consume_envelope

Same as BeeOSAgentRuntime.consume_envelope.

def consume_envelope(envelope: dict)

Python · Parameters

Parameter	Type	Required	Default	Description
`envelope`	`dict`	yes		Message envelope as received.

Returns void

chat.patch_reply

Changes a reply message that Cloud created for a prompt, for example its body, parts or state. Raises BeeOSAgentError with status 400 when a conversation id or reply id is missing.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`conversation_id`	`str`	yes		Identifier of the conversation.
`reply_message_id`	`str`	yes		Identifier of the reply message.
`body`	RuntimeMessagePatch	yes		Request body of the call, as JSON-compatible data.
`operation_key`	`str`	no	`''`	Idempotency key, sent as `Idempotency-Key`. Empty means no key.

Returns void

chat.complete_reply

Finishes the reply as completed with the given text as body and as a single text part, and the stop reason end_turn.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`conversation_id`	`str`	yes		Identifier of the conversation.
`reply_message_id`	`str`	yes		Identifier of the reply message.
`text`	`str`	yes		Final text of the reply.
`operation_key`	`str`	no	`''`	Idempotency key, sent as `Idempotency-Key`. Empty means no key.

Returns void

chat.fail_reply

Finishes the reply as failed with the given body and the stop reason error.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`conversation_id`	`str`	yes		Identifier of the conversation.
`reply_message_id`	`str`	yes		Identifier of the reply message.
`text`	`str`	no	`''`	Body to store with the failed reply, for example an error text. Empty by default.
`operation_key`	`str`	no	`''`	Idempotency key, sent as `Idempotency-Key`. Empty means no key.

Returns void

files.list

Lists the files of the agent.

def list()

Python · Parameters

No parameters.

Returns void

files.presign

Requests a presigned upload for a file.

def presign(input)

Python · Parameters

Parameter	Type	Required	Default	Description
`input`		yes		Presign request body as a dict.

Returns void

files.confirm

Confirms that an uploaded file is complete.

def confirm(input)

Python · Parameters

Parameter	Type	Required	Default	Description
`input`		yes		Confirm request body as a dict.

Returns void

files.resolve

Resolves a file id into its download information.

def resolve(file_id)

Python · Parameters

Parameter	Type	Required	Default	Description
`file_id`		yes		Identifier of the file.

Returns void

canvas.token

Creates a canvas token.

def token(input=None)

Python · Parameters

Parameter	Type	Required	Default	Description
`input`		no	`None`	Optional request body as a dict; an empty body is sent by default.

Returns void

a2a.discover

Discovers agents that can be called through agent-to-agent calls.

def discover()

Python · Parameters

No parameters.

Returns void

a2a.call

Calls another agent with a JSON-RPC request.

def call(agent_id, input)

Python · Parameters

Parameter	Type	Required	Default	Description
`agent_id`		yes		Identifier of the agent to call.
`input`		yes		The JSON-RPC request as a dict.

Returns void

a2a.get_task

Reads the state of an agent-to-agent task.

def get_task(task_id)

Python · Parameters

Parameter	Type	Required	Default	Description
`task_id`		yes		Identifier of the task.

Returns void

messaging.token

Creates a messaging token.

def token(input=None)

Python · Parameters

Parameter	Type	Required	Default	Description
`input`		no	`None`	Optional request body as a dict; an empty body is sent by default.

Returns void

extract_chat_prompt

Python beeos_cloud_harness_sdk · Python only

Extracts a prompt from a message envelope: the text, attached files, conversation id, session key and message id. Returns None when the envelope contains no text.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`envelope`	`dict`	yes		Message envelope as received.
`local_agent_id`	`str`	no	`''`	Identifier of this agent; used to build the session key.

Returns dict \| None

extract_session_prompt

Python beeos_cloud_harness_sdk · Python only

Turns the session/prompt parameters into a chat event. Returns None when the conversation id, reply message id, task request message id or run id is missing, so that the caller can fail the operation instead of writing to an unbound reply.

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

Python · Parameters

Parameter	Type	Required	Default	Description
`params`	`dict`	yes		Parameters of the `session/prompt` call.
`local_agent_id`	`str`	no	`''`	Identifier of this agent, used when the parameters name no agent.

Returns dict \| None

BeeOSAgentError

Python beeos_cloud_harness_sdk · Python only

Error of the facade, carrying the HTTP status and a machine-readable code.

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

Python · Constructor

Parameter	Type	Required	Description
`status`	`int`	yes	HTTP status code of the failure.
`code`	`str`	yes	Machine-readable error code.
`message`	`str`	yes	Error message.

Errors and exceptions

Error types that this module defines. The description of each method says when it raises one.

Errors	Language	Description
BeeOSAgentError	Python	Error of the facade, carrying the HTTP `status` and a machine-readable `code`.

BeeOS CloudBack to documentation