World

Runnable example

python/examples/world_browse.py — session list → join → list open worlds → focus → leave.

resoio.world.WorldClient

WorldClient(socket_path: str | None = None)

Bases: _BaseClient[WorldStub]

Async client for the Resonite IO World service over a UDS.

Use as an async context manager so the gRPC channel closes deterministically.

Source code in src/resoio/_client.py

def __init__(self, socket_path: str | None = None) -> None:
    # Defer resolution to __aenter__ so env vars patched between
    # construction and connection are honoured, and so resolution
    # errors surface at the connect site.
    self._explicit_path: str | None = socket_path
    self._channel: Channel | None = None
    self._stub: TStub | None = None
    self._resolved_path: str | None = None

list_sessions async

list_sessions(
    *,
    search: str = "",
    filter: SessionFilter = SessionFilter.ALL,
    min_active_users: int = 0,
    page: int = 0,
    page_size: int = 0,
) -> ListSessionsResponse

List live sessions (filter / search / paging applied mod-side).

Source code in src/resoio/world.py

async def list_sessions(
    self,
    *,
    search: str = "",
    filter: SessionFilter = SessionFilter.ALL,
    min_active_users: int = 0,
    page: int = 0,
    page_size: int = 0,
) -> ListSessionsResponse:
    """List live sessions (filter / search / paging applied mod-side)."""
    stub = self._require_stub()
    request = ListSessionsRequest(
        search=search,
        filter=_SESSION_FILTER_TO_WIRE[filter],
        min_active_users=min_active_users,
        page=page,
        page_size=page_size,
    )
    return await stub.list_sessions(request)

list_records async

list_records(
    *,
    source: RecordSource = RecordSource.PUBLIC,
    required_tags: Sequence[str] = (),
    owner_id: str = "",
    search: str = "",
    offset: int = 0,
    count: int = 0,
    sort: RecordSort = RecordSort.CREATION_DATE,
    sort_direction: RecordSortDirection = RecordSortDirection.DESCENDING,
) -> ListRecordsResponse

List world records (use sort=RANDOM for the random tab).

search is a free-text query mirroring the World tab (+term required / -term excluded / "phrase"; empty = no search).

Source code in src/resoio/world.py

async def list_records(
    self,
    *,
    source: RecordSource = RecordSource.PUBLIC,
    required_tags: Sequence[str] = (),
    owner_id: str = "",
    search: str = "",
    offset: int = 0,
    count: int = 0,
    sort: RecordSort = RecordSort.CREATION_DATE,
    sort_direction: RecordSortDirection = RecordSortDirection.DESCENDING,
) -> ListRecordsResponse:
    """List world records (use ``sort=RANDOM`` for the random tab).

    ``search`` is a free-text query mirroring the World tab (``+term``
    required / ``-term`` excluded / ``"phrase"``; empty = no search).
    """
    stub = self._require_stub()
    request = ListRecordsRequest(
        source=_RECORD_SOURCE_TO_WIRE[source],
        required_tags=list(required_tags),
        owner_id=owner_id,
        search=search,
        offset=offset,
        count=count,
        sort=_RECORD_SORT_TO_WIRE[sort],
        sort_direction=_RECORD_SORT_DIRECTION_TO_WIRE[sort_direction],
    )
    return await stub.list_records(request)

join async

join(*, session_id: str = '', url: str = '', focus: bool = True) -> OpenWorld

Join an existing session by session_id or by url.

Exactly one of session_id / url must be supplied.

Source code in src/resoio/world.py

async def join(
    self,
    *,
    session_id: str = "",
    url: str = "",
    focus: bool = True,
) -> OpenWorld:
    """Join an existing session by ``session_id`` or by ``url``.

    Exactly one of ``session_id`` / ``url`` must be supplied.
    """
    if bool(session_id) == bool(url):
        raise ValueError("join() requires exactly one of 'session_id' or 'url'.")
    stub = self._require_stub()
    request = JoinRequest(
        session_id=session_id,
        session_url=url,
        focus=focus,
    )
    response = await stub.join(request)
    return _require_world(response.world)

start_world async

start_world(*, record_id: str, owner_id: str = '', focus: bool = True) -> OpenWorld

Start a new session from a world record.

Source code in src/resoio/world.py

async def start_world(
    self,
    *,
    record_id: str,
    owner_id: str = "",
    focus: bool = True,
) -> OpenWorld:
    """Start a new session from a world record."""
    stub = self._require_stub()
    request = StartWorldRequest(
        record_id=record_id,
        owner_id=owner_id,
        focus=focus,
    )
    response = await stub.start_world(request)
    return _require_world(response.world)

list_open_worlds async

list_open_worlds() -> list[OpenWorld]

List the locally-open worlds.

Source code in src/resoio/world.py

async def list_open_worlds(self) -> list[OpenWorld]:
    """List the locally-open worlds."""
    stub = self._require_stub()
    response = await stub.list_open_worlds(ListOpenWorldsRequest())
    return list(response.worlds)

focus async

focus(handle: int) -> OpenWorld

Focus a locally-open world by handle.

Source code in src/resoio/world.py

async def focus(self, handle: int) -> OpenWorld:
    """Focus a locally-open world by handle."""
    stub = self._require_stub()
    response = await stub.focus(FocusRequest(handle=handle))
    return _require_world(response.world)

leave async

leave(handle: int) -> None

Leave a locally-open world by handle.

Source code in src/resoio/world.py

async def leave(self, handle: int) -> None:
    """Leave a locally-open world by handle."""
    stub = self._require_stub()
    await stub.leave(LeaveRequest(handle=handle))

get_current async

get_current() -> OpenWorld | None

Return the currently focused world, or None when in userspace.

Source code in src/resoio/world.py

async def get_current(self) -> OpenWorld | None:
    """Return the currently focused world, or ``None`` when in
    userspace."""
    stub = self._require_stub()
    response = await stub.get_current(GetCurrentRequest())
    if not response.has_world:
        return None
    return response.world

fetch_thumbnail async

fetch_thumbnail(uri: str) -> FetchThumbnailResponse

Fetch a thumbnail image by its resdb:/// or https:// URI.

Source code in src/resoio/world.py

async def fetch_thumbnail(self, uri: str) -> FetchThumbnailResponse:
    """Fetch a thumbnail image by its ``resdb:///`` or ``https://`` URI."""
    stub = self._require_stub()
    return await stub.fetch_thumbnail(FetchThumbnailRequest(uri=uri))

resoio.world.WorldRecord dataclass

WorldRecord(
    record_id: str = betterproto2.field(1, betterproto2.TYPE_STRING),
    owner_id: str = betterproto2.field(2, betterproto2.TYPE_STRING),
    name: str = betterproto2.field(3, betterproto2.TYPE_STRING),
    description: str = betterproto2.field(4, betterproto2.TYPE_STRING),
    thumbnail_url: str = betterproto2.field(5, betterproto2.TYPE_STRING),
    tags: list[str] = betterproto2.field(6, betterproto2.TYPE_STRING, repeated=True),
    record_url: str = betterproto2.field(7, betterproto2.TYPE_STRING),
    last_modification_unix_nanos: int = betterproto2.field(8, betterproto2.TYPE_INT64),
)

Bases: Message

ワールドレコード 1 件の情報 (保存済みワールド / テンプレート)。

thumbnail_url class-attribute instance-attribute

thumbnail_url: str = field(5, TYPE_STRING)

Record のサムネ URI をそのまま載せる (空もありうる)。

record_url class-attribute instance-attribute

record_url: str = field(7, TYPE_STRING)

resrec:// 形式の record URI。

resoio.world.WorldSession dataclass

WorldSession(
    session_id: str = betterproto2.field(1, betterproto2.TYPE_STRING),
    name: str = betterproto2.field(2, betterproto2.TYPE_STRING),
    description: str = betterproto2.field(3, betterproto2.TYPE_STRING),
    host_user_id: str = betterproto2.field(4, betterproto2.TYPE_STRING),
    host_username: str = betterproto2.field(5, betterproto2.TYPE_STRING),
    session_urls: list[str] = betterproto2.field(
        6, betterproto2.TYPE_STRING, repeated=True
    ),
    thumbnail_url: str = betterproto2.field(7, betterproto2.TYPE_STRING),
    joined_users: int = betterproto2.field(8, betterproto2.TYPE_INT32),
    active_users: int = betterproto2.field(9, betterproto2.TYPE_INT32),
    maximum_users: int = betterproto2.field(10, betterproto2.TYPE_INT32),
    tags: list[str] = betterproto2.field(11, betterproto2.TYPE_STRING, repeated=True),
    access_level: str = betterproto2.field(12, betterproto2.TYPE_STRING),
    headless_host: bool = betterproto2.field(13, betterproto2.TYPE_BOOL),
    mobile_friendly: bool = betterproto2.field(14, betterproto2.TYPE_BOOL),
    corresponding_world_id: str = betterproto2.field(15, betterproto2.TYPE_STRING),
    universe_id: str = betterproto2.field(16, betterproto2.TYPE_STRING),
    session_begin_unix_nanos: int = betterproto2.field(17, betterproto2.TYPE_INT64),
    last_update_unix_nanos: int = betterproto2.field(18, betterproto2.TYPE_INT64),
)

Bases: Message

ライブセッション 1 件の情報 (ブラウザのタイル相当)。

session_urls class-attribute instance-attribute

session_urls: list[str] = field(6, TYPE_STRING, repeated=True)

join 先 URI 群 (Join に session_id を渡せば mod 側で解決するため通常は参照のみ)。

thumbnail_url class-attribute instance-attribute

thumbnail_url: str = field(7, TYPE_STRING)

SessionInfo.ThumbnailUrl をそのまま載せる (空もありうる)。

access_level class-attribute instance-attribute

access_level: str = field(12, TYPE_STRING)

SessionAccessLevel の enum 名 (例 "Anyone" / "ContactsPlus")。

resoio.world.OpenWorld dataclass

OpenWorld(
    handle: int = betterproto2.field(1, betterproto2.TYPE_INT32),
    session_id: str = betterproto2.field(2, betterproto2.TYPE_STRING),
    name: str = betterproto2.field(3, betterproto2.TYPE_STRING),
    focused: bool = betterproto2.field(4, betterproto2.TYPE_BOOL),
    user_count: int = betterproto2.field(5, betterproto2.TYPE_INT32),
    access_level: str = betterproto2.field(6, betterproto2.TYPE_STRING),
)

Bases: Message

ローカルに開いているワールド 1 件 (Nest で複数並びうる)。

handle class-attribute instance-attribute

handle: int = field(1, TYPE_INT32)

World.LocalWorldHandle。Focus / Leave の識別子。

focused class-attribute instance-attribute

focused: bool = field(4, TYPE_BOOL)

WorldManager.FocusedWorld と一致するなら true。

resoio.world.ListRecordsResponse dataclass

ListRecordsResponse(
    records: list[WorldRecord] = betterproto2.field(
        1, betterproto2.TYPE_MESSAGE, repeated=True
    ),
    has_more: bool = betterproto2.field(2, betterproto2.TYPE_BOOL),
    offset: int = betterproto2.field(3, betterproto2.TYPE_UINT32),
)

Bases: Message

has_more class-attribute instance-attribute

has_more: bool = field(2, TYPE_BOOL)

まだ続きがあるか。

resoio.world.RecordSort

Bases: Enum

Record search ordering.

resoio.world.RecordSortDirection

Bases: Enum

Record search ordering direction.

resoio.world.RecordSource

Bases: Enum

World-record source (left-tab equivalent).

resoio.world.SessionFilter

Bases: Enum

Live-session filter (left-tab equivalent).

ALL = no filter.

resoio.world.ListSessionsResponse dataclass

ListSessionsResponse(
    sessions: list[WorldSession] = betterproto2.field(
        1, betterproto2.TYPE_MESSAGE, repeated=True
    ),
    total_count: int = betterproto2.field(2, betterproto2.TYPE_UINT32),
    page: int = betterproto2.field(3, betterproto2.TYPE_UINT32),
    page_size: int = betterproto2.field(4, betterproto2.TYPE_UINT32),
)

Bases: Message

total_count class-attribute instance-attribute

total_count: int = field(2, TYPE_UINT32)

ページング前の総件数。

resoio.world.FetchThumbnailResponse dataclass

FetchThumbnailResponse(
    data: bytes = betterproto2.field(1, betterproto2.TYPE_BYTES),
    content_type: str = betterproto2.field(2, betterproto2.TYPE_STRING),
)

Bases: Message

content_type class-attribute instance-attribute

content_type: str = field(2, TYPE_STRING)

MIME (例 "image/webp")。判定不能なら空。