类class agentscope.app.message_bus

MessageBus

class MessageBus(ABC)

源码Source agentscope/app/message_bus/_base.py:53–910

消息总线，把智能体接入外部消息渠道与企业平台。位于 agentscope.app.message_bus 包，继承自 ABC。核心成员包括 aclose、queue_push、queue_drain、queue_delete、log_append、log_read。

Implementations expose three consumption modes (drain queue, replay log, transient broadcast) over arbitrary string keys and JSON-style dict payloads. Callers own key naming and payload schemas.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 Open underlying transport resources (connection pools, …).
`async def __aexit__(exc_type: type[BaseException] \| None, exc_value: BaseException \| None, traceback: Any) -> None`async	退出异步上下文并清理资源。 Release transport resources on context exit.
`async def aclose() -> None`async	返回 `aclose` 结果。 Release underlying transport resources. Default is a no-op.
`async def queue_push(key: str, payload: dict, *, ttl_secs: int \| None = None) -> str`asyncabstractabstractmethod	执行队列推送相关操作。 Append ``payload`` to the drain queue at ``key``.
`async def queue_drain(key: str, max_count: int = 100) -> list[tuple[str, dict]]`asyncabstractabstractmethod	执行 `queue_drain` 方法。 Drain up to ``max_count`` entries from the queue at ``key``.
`async def queue_delete(key: str) -> None`asyncabstractabstractmethod	设置队列删除。 Delete the drain queue at ``key`` and all of its entries.
`async def log_append(key: str, payload: dict, *, ttl_secs: int \| None = None, max_len: int \| None = None) -> str`asyncabstractabstractmethod	执行日志追加相关操作。 Append ``payload`` to the replay log at ``key``.
`async def log_read(key: str, since: str \| None = None, max_count: int = 100) -> list[tuple[str, dict]]`asyncabstractabstractmethod	执行日志读取相关操作。 Read up to ``max_count`` entries from the replay log at ``key``, starting after ``since``.
`async def log_trim(key: str, before_id: str \| None = None) -> None`asyncabstractabstractmethod	执行日志裁剪相关操作。 Trim the replay log at ``key``.
`async def publish(key: str, payload: dict) -> None`asyncabstractabstractmethod	执行发布相关操作。 Publish ``payload`` on the broadcast channel ``key``.
`async def subscribe(key: str, *, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[dict, None]`asyncabstractabstractmethod	执行订阅相关操作。 Yield broadcast payloads for ``key`` until the consumer closes the generator.
`async def acquire_lock(key: str, *, ttl_secs: int = 600) -> AsyncGenerator[None, None]`asyncabstractabstractmethodasynccontextmanager	执行 `acquire_lock` 方法。 Acquire a distributed mutex on ``key``.
`async def is_locked(key: str) -> bool`asyncabstractabstractmethod	判断是否 `is_locked`。 Return whether ``key`` currently holds a lock.
`async def try_lock(key: str, *, ttl_secs: int = 600) -> bool`asyncabstractabstractmethod	执行 `try_lock` 方法。 Claim ``key`` without blocking (non-blocking mutex).
`async def unlock(key: str) -> None`asyncabstractabstractmethod	设置该参数。 Release a lock claimed via :meth:`try_lock` (best-effort).
`async def registry_set(namespace: str, field: str, value: str, *, ttl_secs: int \| None = None) -> None`asyncabstractabstractmethod	设置`field`到`value`in 注册表 at`namespace`。 Set ``field`` to ``value`` in the registry at ``namespace``.
`async def registry_set_if(namespace: str, field: str, value: str, *, expected: str, ttl_secs: int \| None = None) -> bool`asyncabstractabstractmethod	执行注册表设置 if相关操作。 Set ``field`` only while it still holds ``expected``.
`async def registry_pop(namespace: str, field: str) -> str \| None`asyncabstractabstractmethod	执行 `registry_pop` 方法。 Read ``field`` and remove it in one step.
`async def registry_del(namespace: str, field: str) -> None`asyncabstractabstractmethod	执行 `registry_del` 方法。 Remove ``field`` from the registry at ``namespace``.
`async def registry_exists(namespace: str, field: str) -> bool`asyncabstractabstractmethod	执行 `registry_exists` 方法。 Return whether ``field`` exists in the registry at ``namespace``.
`async def registry_getall(namespace: str) -> dict[str, str]`asyncabstractabstractmethod	设置该参数。 Return all field-value pairs in the registry at ``namespace``.
`async def registry_get(namespace: str, field: str) -> str \| None`asyncabstractabstractmethod	执行注册表读取相关操作。 Return the value of a single ``field`` in the registry at ``namespace``, or ``None`` if absent.
`async def registry_drop(namespace: str) -> None`asyncabstractabstractmethod	设置该参数。 Delete the entire registry at ``namespace``.
`async def session_run(session_id: str) -> AsyncGenerator[None, None]`asyncdeprecateddeprecatedasynccontextmanager	设置会话运行。 Acquire the session lock, yield, then trim the replay log.
`async def session_is_running(session_id: str) -> bool`asyncdeprecateddeprecated	设置会话是运行中。 Check whether some process holds the session lock.
`async def session_publish_event(session_id: str, event: dict) -> str`asyncdeprecateddeprecated	执行会话发布事件相关操作。 Append + fan-out a session event.
`async def session_read_events(session_id: str, since: str \| None = None, max_count: int = 1000) -> list[tuple[str, dict…`asyncdeprecateddeprecated	执行会话读取事件相关操作。 Read events from the session's replay log.
`async def session_subscribe_events(session_id: str, *, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[di…`asyncdeprecateddeprecated	执行会话订阅事件相关操作。 Live-subscribe to session events, stripping _entry_id.
`async def session_publish_cancel(session_id: str) -> None`asyncdeprecateddeprecated	设置会话发布取消。 Broadcast a session cancel request.
`async def session_subscribe_cancel(*, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[str, None]`asyncdeprecateddeprecated	设置会话订阅取消。 Subscribe to session cancel broadcasts, yielding session ids.
`async def session_purge(session_id: str) -> None`asyncdeprecateddeprecated	设置该参数。 Delete all per-session bus state.
`async def inbox_push(session_id: str, msg: dict, *, ttl_secs: int \| None = None) -> str`asyncdeprecateddeprecated	执行收件箱推送相关操作。 Push a message to a session's inbox.
`async def inbox_drain(session_id: str, max_count: int = 100) -> list[tuple[str, dict]]`asyncdeprecateddeprecated	执行 `inbox_drain` 方法。 Drain pending inbox messages for a session.
`async def enqueue_wakeup(user_id: str, session_id: str, agent_id: str) -> None`asyncdeprecateddeprecated	执行 `enqueue_wakeup` 方法。 Enqueue an idle-session wake-up.
`async def enqueue_input(user_id: str, session_id: str, agent_id: str, *, kind: str, inputs: dict \| None = None) -> None`asyncdeprecateddeprecated	执行 `enqueue_input` 方法。 Enqueue a typed run trigger.
`async def dequeue_wakeups(max_count: int = 64) -> list[dict]`asyncdeprecateddeprecated	设置该参数。 Drain pending run-trigger entries.
`async def subscribe_wakeup_signal(*, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[dict, None]`asyncdeprecateddeprecated	订阅 `subscribe_wakeup_signal`。 Subscribe to the shared wake-up signal channel.
`async def bg_task_register(session_id: str, task_id: str, metadata: str) -> None`asyncdeprecateddeprecated	执行bg 任务注册相关操作。 Register a background task.
`async def bg_task_unregister(session_id: str, task_id: str) -> None`asyncdeprecateddeprecated	执行bg 任务注销相关操作。 Unregister a background task.
`async def bg_task_exists(session_id: str, task_id: str) -> bool`asyncdeprecateddeprecated	执行 `bg_task_exists` 方法。 Check whether a background task is registered.
`async def bg_task_list(session_id: str) -> dict[str, str]`asyncdeprecateddeprecated	设置bg 任务列表。 List all background tasks for a session.
`async def bg_task_purge(session_id: str) -> None`asyncdeprecateddeprecated	设置该参数。 Delete all background task entries for a session.
`async def task_publish_cancel(task_id: str) -> None`asyncdeprecateddeprecated	设置任务发布取消。 Broadcast a cancel request for a single background task.
`async def task_subscribe_cancel(*, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[str, None]`asyncdeprecateddeprecated	设置任务订阅取消。 Subscribe to task cancel broadcasts, yielding task ids.

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

进入异步上下文并返回自身。

Open underlying transport resources (connection pools, …).

返回Returns Self

返回自身。`Self`:

源码Source agentscope/app/message_bus/_base.py:61–68

aexit

async def __aexit__(exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: Any) -> None

async

退出异步上下文并清理资源。

Release transport resources on context exit.

参数Parameters

名称Name	类型Type	说明Description
`exc_type`	type[BaseException] \| None	exc 类型（类型 `type[BaseException] \| None`）。The exception class raised inside the context, if any.
`exc_value`	BaseException \| None	exc 值（类型 `BaseException \| None`）。The exception instance raised inside the context, if any.
`traceback`	Any	traceback（类型 `Any`）。The traceback associated with the exception, if any.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:70–86

aclose

async def aclose() -> None

async

返回 aclose 结果。

Release underlying transport resources.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:88–89

queue_push

async def queue_push(key: str, payload: dict, *, ttl_secs: int | None = None) -> str

asyncabstractabstractmethod

执行队列推送相关操作。

Drain queues are ack-on-read: :meth:queue_drain returns each entry to exactly one caller and then deletes it.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Queue identifier. Caller-defined naming convention; the bus treats it as opaque.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict to enqueue. Schema is the caller's responsibility.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。If set, refresh the queue key's expiry to this many seconds on every push (sliding TTL). When `None`, the key never expires and must be drained or deleted explicitly.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/message_bus/_base.py:96–128

queue_drain

async def queue_drain(key: str, max_count: int = 100) -> list[tuple[str, dict]]

asyncabstractabstractmethod

执行 queue_drain 方法。

Reading and removing are one atomic operation, so concurrent callers on the same key get disjoint entries and each entry is delivered once.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Queue identifier.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum number of entries to return in one call. Older entries are returned first; remaining entries stay in the queue for the next call.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。`list[tuple[str, dict]]`:

源码Source agentscope/app/message_bus/_base.py:131–156

queue_delete

async def queue_delete(key: str) -> None

asyncabstractabstractmethod

设置队列删除。

Idempotent: a no-op when the key does not exist.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Queue identifier.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:159–169

log_append

async def log_append(key: str, payload: dict, *, ttl_secs: int | None = None, max_len: int | None = None) -> str

asyncabstractabstractmethod

执行日志追加相关操作。

Replay logs are append-only; readers track their own cursor and may join at any time.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Log identifier.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict to append.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。If set, refresh the key's expiry to this many seconds on every append (sliding TTL). `None` means no TTL.
`max_len`kw默认default `None`	int \| None	最大 len（类型 `int \| None`）。If set, cap the log at approximately this many entries — older entries are trimmed when the cap is exceeded. The cap is approximate so the operation can use the backend's efficient near-trim mode (e.g. `XADD MAXLEN ~N`). `None` means no cap.

返回Returns str

返回字符串结果。`str`:

源码Source agentscope/app/message_bus/_base.py:176–211

log_read

async def log_read(key: str, since: str | None = None, max_count: int = 100) -> list[tuple[str, dict]]

asyncabstractabstractmethod

执行日志读取相关操作。

Reads are non-destructive: the same entries can be returned to any number of readers, or to the same reader on retry.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Log identifier.
`since`默认default `None`	str \| None	since（类型 `str \| None`）。Cursor — return entries strictly newer than this id. Pass the last `entry_id` from the previous read. `None` reads from the beginning of the log.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum number of entries to return.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。`list[tuple[str, dict]]`:

源码Source agentscope/app/message_bus/_base.py:214–241

log_trim

async def log_trim(key: str, before_id: str | None = None) -> None

asyncabstractabstractmethod

执行日志裁剪相关操作。

Trim the replay log at key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Log identifier.
`before_id`默认default `None`	str \| None	之前标识（类型 `str \| None`）。Drop all entries with id strictly older than this. `None` drops the entire log (i.e. deletes the key).

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:244–257

publish

async def publish(key: str, payload: dict) -> None

asyncabstractabstractmethod

执行发布相关操作。

Only subscribers connected at the moment of publish receive the payload — no history is retained.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Channel identifier.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict delivered as-is to subscribers.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:264–281

async def subscribe(key: str, *, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[dict, None]

asyncabstractabstractmethod

执行订阅相关操作。

Subscriptions are best-effort: only payloads published after the subscription is established are delivered.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Channel identifier.
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。If supplied, invoked exactly once after the underlying subscription is established and before any payload is yielded. Used by callers that need to block a bootstrapping step (e.g. `SessionTriggerListenerManager.start`) until the subscription is live, so a publish-immediately-after race is impossible.

返回Returns AsyncGenerator[dict, None]

以异步流的形式返回dict,None。`dict`:

源码Source agentscope/app/message_bus/_base.py:284–319

acquire_lock

async def acquire_lock(key: str, *, ttl_secs: int = 600) -> AsyncGenerator[None, None]

asyncabstractabstractmethodasynccontextmanager

执行 acquire_lock 方法。

Blocks until the lock is acquired, then yields.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。Lease duration in seconds. The implementation should renew this periodically while the body runs; callers do not need to.

返回Returns AsyncGenerator[None, None]

以异步流的形式返回None,None。`None`: while the lock is held.

源码Source agentscope/app/message_bus/_base.py:327–356

is_locked

async def is_locked(key: str) -> bool

asyncabstractabstractmethod

判断是否 is_locked。

Return whether key currently holds a lock.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier (same key passed to :meth:`acquire_lock`).

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_base.py:359–370

try_lock

async def try_lock(key: str, *, ttl_secs: int = 600) -> bool

asyncabstractabstractmethod

执行 try_lock 方法。

Unlike :meth:acquire_lock (which waits until free), this returns immediately: True if acquired, False if already held.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。Lease duration; the claim expires automatically so a crashed holder cannot block the key forever.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_base.py:373–392

unlock

async def unlock(key: str) -> None

asyncabstractabstractmethod

设置该参数。

Release a lock claimed via :meth:try_lock (best-effort).

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。The key.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:395–396

registry_set

async def registry_set(namespace: str, field: str, value: str, *, ttl_secs: int | None = None) -> None

asyncabstractabstractmethod

设置field到valuein 注册表 atnamespace。

If the namespace does not exist it is created.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key (e.g. `"agentscope:bg_tasks:sess123"`).
`field`	str	field（类型 `str`）。Field name within the registry.
`value`	str	值（类型 `str`）。Serialized value to store.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。Refresh the namespace expiry to this many seconds.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:403–426

registry_set_if

async def registry_set_if(namespace: str, field: str, value: str, *, expected: str, ttl_secs: int | None = None) -> bool

asyncabstractabstractmethod

执行注册表设置 if相关操作。

Compare-and-set, so a caller that read a value, decided on a new one, and comes back to write it can tell whether anyone else moved it meanwhile — without holding a lock across the think time.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field name within the registry.
`value`	str	值（类型 `str`）。Serialized value to store.
`expected`kw	str	expected（类型 `str`）。The value the caller last read. The write is skipped when the stored value differs.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。Refresh the namespace expiry when the write happens.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_base.py:429–461

registry_pop

async def registry_pop(namespace: str, field: str) -> str | None

asyncabstractabstractmethod

执行 registry_pop 方法。

Lets a value be consumed exactly once however many callers race for it: only one sees it, the rest get None.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field to take.

返回Returns str | None

返回str|None。`str | None`:

源码Source agentscope/app/message_bus/_base.py:464–479

registry_del

async def registry_del(namespace: str, field: str) -> None

asyncabstractabstractmethod

执行 registry_del 方法。

A no-op when the field or namespace does not exist.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field to remove.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:482–492

registry_exists

async def registry_exists(namespace: str, field: str) -> bool

asyncabstractabstractmethod

执行 registry_exists 方法。

Return whether field exists in the registry at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field to check.

返回Returns bool

返回布尔值，表示判断结果。`bool`:

源码Source agentscope/app/message_bus/_base.py:495–508

registry_getall

async def registry_getall(namespace: str) -> dict[str, str]

asyncabstractabstractmethod

设置该参数。

Return all field-value pairs in the registry at namespace.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.

返回Returns dict[str, str]

返回dict[str,str]。`dict[str, str]`:

源码Source agentscope/app/message_bus/_base.py:511–525

registry_get

async def registry_get(namespace: str, field: str) -> str | None

asyncabstractabstractmethod

执行注册表读取相关操作。

Return the value of a single field in the registry at namespace, or None if absent.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field to retrieve.

返回Returns str | None

返回str|None。`str | None`:

源码Source agentscope/app/message_bus/_base.py:528–545

registry_drop

async def registry_drop(namespace: str) -> None

asyncabstractabstractmethod

设置该参数。

Idempotent: a no-op when the namespace does not exist.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key to delete.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:548–556

session_run

async def session_run(session_id: str) -> AsyncGenerator[None, None]

asyncdeprecateddeprecatedasynccontextmanager

设置会话运行。

Acquire the session lock, yield, then trim the replay log.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns AsyncGenerator[None, None]

以异步流的形式返回None,None。Returns a stream of none none.

源码Source agentscope/app/message_bus/_base.py:590–601

session_is_running

async def session_is_running(session_id: str) -> bool

asyncdeprecateddeprecated

设置会话是运行中。

Check whether some process holds the session lock.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/app/message_bus/_base.py:606–610

session_publish_event

async def session_publish_event(session_id: str, event: dict) -> str

asyncdeprecateddeprecated

执行会话发布事件相关操作。

Append + fan-out a session event.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`event`	dict	事件（类型 `dict`）。The event.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/app/message_bus/_base.py:616–629

session_read_events

async def session_read_events(session_id: str, since: str | None = None, max_count: int = 1000) -> list[tuple[str, dict]]

asyncdeprecateddeprecated

执行会话读取事件相关操作。

Read events from the session's replay log.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`since`默认default `None`	str \| None	since（类型 `str \| None`）。The since.
`max_count`默认default `1000`	int	最大数量（类型 `int`）。The max count.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。Returns a list of tuple str dict.

源码Source agentscope/app/message_bus/_base.py:634–645

session_subscribe_events

async def session_subscribe_events(session_id: str, *, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[dict, None]

asyncdeprecateddeprecated

执行会话订阅事件相关操作。

Live-subscribe to session events, stripping _entry_id.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。The on ready.

返回Returns AsyncGenerator[dict, None]

以异步流的形式返回dict,None。Returns a stream of dict none.

源码Source agentscope/app/message_bus/_base.py:651–660

session_publish_cancel

async def session_publish_cancel(session_id: str) -> None

asyncdeprecateddeprecated

设置会话发布取消。

Broadcast a session cancel request.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:668–673

session_subscribe_cancel

async def session_subscribe_cancel(*, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[str, None]

asyncdeprecateddeprecated

设置会话订阅取消。

Subscribe to session cancel broadcasts, yielding session ids.

参数Parameters

名称Name	类型Type	说明Description
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。The on ready.

返回Returns AsyncGenerator[str, None]

以异步流的形式返回str,None。Returns a stream of str none.

源码Source agentscope/app/message_bus/_base.py:679–691

session_purge

async def session_purge(session_id: str) -> None

asyncdeprecateddeprecated

设置该参数。

Delete all per-session bus state.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:699–703

inbox_push

async def inbox_push(session_id: str, msg: dict, *, ttl_secs: int | None = None) -> str

asyncdeprecateddeprecated

执行收件箱推送相关操作。

Push a message to a session's inbox.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`msg`	dict	消息（类型 `dict`）。The msg.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。The ttl secs.

返回Returns str

返回字符串结果。Returns a string.

源码Source agentscope/app/message_bus/_base.py:710–722

inbox_drain

async def inbox_drain(session_id: str, max_count: int = 100) -> list[tuple[str, dict]]

asyncdeprecateddeprecated

执行 inbox_drain 方法。

Drain pending inbox messages for a session.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`max_count`默认default `100`	int	最大数量（类型 `int`）。The max count.

返回Returns list[tuple[str, dict]]

返回元组[str,dict]列表。Returns a list of tuple str dict.

源码Source agentscope/app/message_bus/_base.py:727–736

enqueue_wakeup

async def enqueue_wakeup(user_id: str, session_id: str, agent_id: str) -> None

asyncdeprecateddeprecated

执行 enqueue_wakeup 方法。

Enqueue an idle-session wake-up.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:744–761

enqueue_input

async def enqueue_input(user_id: str, session_id: str, agent_id: str, *, kind: str, inputs: dict | None = None) -> None

asyncdeprecateddeprecated

执行 enqueue_input 方法。

Enqueue a typed run trigger.

参数Parameters

名称Name	类型Type	说明Description
`user_id`	str	用户标识（类型 `str`）。The user id.
`session_id`	str	会话标识（类型 `str`）。The session id.
`agent_id`	str	智能体标识（类型 `str`）。The agent id.
`kind`kw	str	种类（类型 `str`）。The kind.
`inputs`kw默认default `None`	dict \| None	输入消息（类型 `dict \| None`）。The inputs.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:767–787

dequeue_wakeups

async def dequeue_wakeups(max_count: int = 64) -> list[dict]

asyncdeprecateddeprecated

设置该参数。

Drain pending run-trigger entries.

参数Parameters

名称Name	类型Type	说明Description
`max_count`默认default `64`	int	最大数量（类型 `int`）。The max count.

返回Returns list[dict]

返回dict列表。Returns a list of dict.

源码Source agentscope/app/message_bus/_base.py:792–801

subscribe_wakeup_signal

async def subscribe_wakeup_signal(*, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[dict, None]

asyncdeprecateddeprecated

订阅 subscribe_wakeup_signal。

Subscribe to the shared wake-up signal channel.

参数Parameters

名称Name	类型Type	说明Description
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。The on ready.

返回Returns AsyncGenerator[dict, None]

以异步流的形式返回dict,None。Returns a stream of dict none.

源码Source agentscope/app/message_bus/_base.py:806–816

bg_task_register

async def bg_task_register(session_id: str, task_id: str, metadata: str) -> None

asyncdeprecateddeprecated

执行bg 任务注册相关操作。

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`task_id`	str	任务标识（类型 `str`）。The task id.
`metadata`	str	元数据（类型 `str`）。The metadata.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:823–835

bg_task_unregister

async def bg_task_unregister(session_id: str, task_id: str) -> None

asyncdeprecateddeprecated

执行bg 任务注销相关操作。

Unregister a background task.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`task_id`	str	任务标识（类型 `str`）。The task id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:840–849

bg_task_exists

async def bg_task_exists(session_id: str, task_id: str) -> bool

asyncdeprecateddeprecated

执行 bg_task_exists 方法。

Check whether a background task is registered.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.
`task_id`	str	任务标识（类型 `str`）。The task id.

返回Returns bool

返回布尔值，表示判断结果。Returns a boolean result.

源码Source agentscope/app/message_bus/_base.py:854–863

bg_task_list

async def bg_task_list(session_id: str) -> dict[str, str]

asyncdeprecateddeprecated

设置bg 任务列表。

List all background tasks for a session.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns dict[str, str]

返回dict[str,str]。Returns dict str str.

源码Source agentscope/app/message_bus/_base.py:868–875

bg_task_purge

async def bg_task_purge(session_id: str) -> None

asyncdeprecateddeprecated

设置该参数。

Delete all background task entries for a session.

参数Parameters

名称Name	类型Type	说明Description
`session_id`	str	会话标识（类型 `str`）。The session id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:880–884

task_publish_cancel

async def task_publish_cancel(task_id: str) -> None

asyncdeprecateddeprecated

设置任务发布取消。

Broadcast a cancel request for a single background task.

参数Parameters

名称Name	类型Type	说明Description
`task_id`	str	任务标识（类型 `str`）。The task id.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_base.py:890–892

task_subscribe_cancel

async def task_subscribe_cancel(*, on_ready: Callable[[], None] | None = None) -> AsyncGenerator[str, None]

asyncdeprecateddeprecated

设置任务订阅取消。

Subscribe to task cancel broadcasts, yielding task ids.

参数Parameters

名称Name	类型Type	说明Description
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。The on ready.

返回Returns AsyncGenerator[str, None]

以异步流的形式返回str,None。Returns a stream of str none.

源码Source agentscope/app/message_bus/_base.py:898–910