Axon

AsyncSync

Axon resource class for asynchronous operations.

class runloop_api_client.sdk.async_axon.AsyncAxonSqlOps(client, axon_id)

[Beta] Async SQL operations for an axon’s SQLite database.

Access via axon.sql.

Example

>>> axon = await runloop.axon.create()
>>> await axon.sql.query(sql="CREATE TABLE tasks (id INTEGER PRIMARY KEY, name TEXT)")
>>> result = await axon.sql.query(sql="SELECT * FROM tasks WHERE id = ?", params=[1])

async query(**params)

[Beta] Execute a single parameterized SQL statement against this axon’s SQLite database.

Return type:

SqlQueryResultView

async batch(**params)

[Beta] Execute multiple SQL statements atomically within a single transaction.

Return type:

SqlBatchResultView

class runloop_api_client.sdk.async_axon.AsyncAxon(client, axon_id)

[Beta] Wrapper around asynchronous axon operations.

Axons are event communication channels that support publishing events, subscribing to event streams via server-sent events (SSE), and executing SQL queries against an embedded SQLite database. Obtain instances via runloop.axon.create() or runloop.axon.from_id().

Example

>>> runloop = AsyncRunloopSDK()
>>> axon = await runloop.axon.create()
>>> await axon.publish(event_type="task_done", origin="AGENT_EVENT", payload="{}", source="my-agent")
>>> async with await axon.subscribe_sse() as stream:
...     async for event in stream:
...         print(event.event_type, event.payload)
>>> await axon.sql.query(sql="CREATE TABLE tasks (id INTEGER PRIMARY KEY, name TEXT)")

async get_info(**options)

[Beta] Retrieve the latest axon information.

Return type:

AxonView

async publish(**params)

[Beta] Publish an event to this axon.

Return type:

PublishResultView

async subscribe_sse(**options)

[Beta] Subscribe to this axon’s event stream via SSE.

Return type:

AsyncStream[AxonEventView]