类class agentscope.app.message_bus

InMemoryMessageBus

class InMemoryMessageBus(MessageBus)

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:30–538

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

Mapping of bus modes to in-memory structures:

Mode A (drain queue) — each key maps to a :class:list[tuple[str, dict]] of (entry_id, payload) pairs. queue_push appends; queue_drain pops from the front (FIFO) and deletes the returned entries.
Mode C (replay log) — same underlying list structure, but log_read is non-destructive. log_trim removes entries in-place.
Mode D (transient broadcast) — each channel keeps a set of :class:asyncio.Queue subscribers. publish pushes to all currently-subscribed queues; subscribe yields from one.
Mode E (distributed lock) — :class:asyncio.Lock per key, suitable only for single-process concurrency.
Mode F (registry map) — dict[str, dict[str, str]], one nested dict per namespace.

Entry ids are monotonic "<seq>-0" strings (e.g. "1-0", "2-0", …). They are not lexicographically sortable once the sequence exceeds single digits; comparison must parse the numeric prefix (as :meth:log_read does internally).

构造器摘要Constructor Summary

构造器Constructor	说明Description
`def __init__() -> None`	初始化 `InMemoryMessageBus` 实例。 Initialise empty in-memory stores.

方法摘要Method Summary

方法Method	说明Description
`async def __aenter__() -> Self`async	进入异步上下文并返回自身。 No-op — nothing to open for in-memory transport.
`async def aclose() -> None`async	返回 `aclose` 结果。 Signal all open subscribers so their generators terminate.
`async def queue_push(key: str, payload: dict, *, ttl_secs: int \| None = None) -> str`async	执行队列推送相关操作。 Append ``payload`` to the in-memory drain queue at ``key``.
`async def queue_drain(key: str, max_count: int = 100) -> list[tuple[str, dict]]`async	执行 `queue_drain` 方法。 Drain up to ``max_count`` entries from the queue at ``key``.
`async def queue_delete(key: str) -> None`async	设置队列删除。 Delete the drain queue at ``key``.
`async def log_append(key: str, payload: dict, *, ttl_secs: int \| None = None, max_len: int \| None = None) -> str`async	执行日志追加相关操作。 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]]`async	执行日志读取相关操作。 Read up to ``max_count`` entries newer than ``since``.
`async def log_trim(key: str, before_id: str \| None = None) -> None`async	执行日志裁剪相关操作。 Trim the replay log at ``key``.
`async def publish(key: str, payload: dict) -> None`async	执行发布相关操作。 Publish ``payload`` to all current subscribers of ``key``.
`async def subscribe(key: str, *, on_ready: Callable[[], None] \| None = None) -> AsyncGenerator[dict, None]`async	执行订阅相关操作。 Yield broadcast payloads for ``key`` until the consumer closes the generator.
`async def acquire_lock(key: str, *, ttl_secs: int = 600) -> AsyncGenerator[None, None]`asyncasynccontextmanager	执行 `acquire_lock` 方法。 Acquire a process-local mutex on ``key``.
`async def is_locked(key: str) -> bool`async	判断是否 `is_locked`。 Return whether ``key`` currently holds a lock.
`async def try_lock(key: str, *, ttl_secs: int = 600) -> bool`async	执行 `try_lock` 方法。 Non-blocking claim on ``key``. See base.
`async def unlock(key: str) -> None`async	设置该参数。 Release a ``try_lock`` claim.
`async def registry_set(namespace: str, field: str, value: str, *, ttl_secs: int \| None = None) -> None`async	设置`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`async	执行注册表设置 if相关操作。 Compare-and-set one field; ``ttl_secs`` ignored. See base.
`async def registry_pop(namespace: str, field: str) -> str \| None`async	执行 `registry_pop` 方法。 Read and remove one field. See base.
`async def registry_del(namespace: str, field: str) -> None`async	执行 `registry_del` 方法。 Remove ``field`` from the registry at ``namespace``.
`async def registry_exists(namespace: str, field: str) -> bool`async	执行 `registry_exists` 方法。 Return whether ``field`` exists in the registry at ``namespace``.
`async def registry_getall(namespace: str) -> dict[str, str]`async	设置该参数。 Return all field-value pairs in the registry at ``namespace``.
`async def registry_get(namespace: str, field: str) -> str \| None`async	执行注册表读取相关操作。 Return a single field value from the registry at ``namespace``, or ``None`` if absent.
`async def registry_drop(namespace: str) -> None`async	设置该参数。 Delete the entire registry at ``namespace``.

构造器详情Constructor Detail

InMemoryMessageBus

def __init__() -> None

初始化 InMemoryMessageBus 实例。

Initialise empty in-memory stores.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:58–87

方法详情Method Detail

aenter

async def __aenter__() -> Self

async

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

No-op — nothing to open for in-memory transport.

返回Returns Self

返回自身。`Self`:

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:93–100

aclose

async def aclose() -> None

async

返回 aclose 结果。

Signal all open subscribers so their generators terminate.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:102–107

queue_push

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

async

执行队列推送相关操作。

When ttl_secs is set the entry expires after that many seconds; expired entries are dropped here and on drain, so a never-drained key stays bounded.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Queue identifier.
`payload`	dict	负载（类型 `dict`）。JSON-serializable dict to enqueue.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。Entry lifetime in seconds; `None` never expires.

返回Returns str

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:130–161

queue_drain

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

async

执行 queue_drain 方法。

Returned entries are removed from the internal list.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Queue identifier.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum entries to return.

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

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:163–189

queue_delete

async def queue_delete(key: str) -> None

async

设置队列删除。

Delete the drain queue at key.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:191–198

log_append

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

async

执行日志追加相关操作。

Append payload to the replay log at key.

参数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`）。Ignored (no-op — in-memory does not expire keys).
`max_len`kw默认default `None`	int \| None	最大 len（类型 `int \| None`）。If set, trim the log to approximately this many entries after appending.

返回Returns str

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:204–234

log_read

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

async

执行日志读取相关操作。

Reads are non-destructive.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Log identifier.
`since`默认default `None`	str \| None	since（类型 `str \| None`）。Exclusive cursor — return entries whose id is strictly greater than `since`. `None` reads from the start.
`max_count`默认default `100`	int	最大数量（类型 `int`）。Maximum entries to return.

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

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:236–276

log_trim

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

async

执行日志裁剪相关操作。

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.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:278–301

publish

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

async

执行发布相关操作。

Publish payload to all current subscribers of key.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:307–321

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

async

执行订阅相关操作。

Yield broadcast payloads for key until the consumer closes the generator.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Channel identifier.
`on_ready`kw默认default `None`	Callable[[], None] \| None	on 就绪（类型 `Callable[[], None] \| None`）。Called once after the subscription is established and before any payload is yielded.

返回Returns AsyncGenerator[dict, None]

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:323–355

acquire_lock

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

asyncasynccontextmanager

执行 acquire_lock 方法。

In-memory equivalent of a distributed lock.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。Lock identifier.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。Ignored (no automatic expiry).

返回Returns AsyncGenerator[None, None]

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:362–389

is_locked

async def is_locked(key: str) -> bool

async

判断是否 is_locked。

Return whether key currently holds a lock.

参数Parameters

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

返回Returns bool

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:391–402

try_lock

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

async

执行 try_lock 方法。

Non-blocking claim on key.

参数Parameters

名称Name	类型Type	说明Description
`key`	str	键（类型 `str`）。The key.
`ttl_secs`kw默认default `600`	int	存活时间 secs（类型 `int`）。The ttl secs.

返回Returns bool

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:404–411

unlock

async def unlock(key: str) -> None

async

设置该参数。

Release a try_lock claim.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:413–415

registry_set

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

async

设置field到valuein 注册表 atnamespace。

ttl_secs is accepted for API compatibility but ignored.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。Registry key.
`field`	str	field（类型 `str`）。Field name.
`value`	str	值（类型 `str`）。Value to store.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。Ignored (no TTL support).

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:421–443

registry_set_if

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

async

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

Compare-and-set one field; ttl_secs ignored.

参数Parameters

名称Name	类型Type	说明Description
`namespace`	str	命名空间（类型 `str`）。The namespace.
`field`	str	field（类型 `str`）。The field.
`value`	str	值（类型 `str`）。The value.
`expected`kw	str	expected（类型 `str`）。The expected.
`ttl_secs`kw默认default `None`	int \| None	存活时间 secs（类型 `int \| None`）。The ttl secs.

返回Returns bool

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

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:445–459

registry_pop

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

async

执行 registry_pop 方法。

Read and remove one field.

参数Parameters

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

返回Returns str | None

返回str|None。Returns str none.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:461–463

registry_del

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

async

执行 registry_del 方法。

Remove field from the registry at namespace.

参数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/_in_memory_message_bus.py:465–476

registry_exists

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

async

执行 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/_in_memory_message_bus.py:478–492

registry_getall

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

async

设置该参数。

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/_in_memory_message_bus.py:494–509

registry_get

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

async

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

Return a single field value from 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/_in_memory_message_bus.py:511–529

registry_drop

async def registry_drop(namespace: str) -> None

async

设置该参数。

Delete the entire registry at namespace.

参数Parameters

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

返回Returns None

无返回值。No value is returned.

源码Source agentscope/app/message_bus/_in_memory_message_bus.py:531–538

InMemoryMessageBus

构造器摘要Constructor Summary

方法摘要Method Summary

构造器详情Constructor Detail

InMemoryMessageBus

方法详情Method Detail

__aenter__

aclose

queue_push

queue_drain

queue_delete

log_append

log_read

log_trim

publish

subscribe

acquire_lock

is_locked

try_lock

unlock

registry_set

registry_set_if

registry_pop

registry_del

registry_exists

registry_getall

registry_get

registry_drop

继承成员Inherited Members

aenter