# Generated from soulfire/bot_live.proto. DO NOT EDIT.
# Generated by protoc-gen-connectrpc-py v0.11.1 with parameter "protobuf=google".
# ruff: noqa: PGH004
# ruff: noqa
# fmt: off
from __future__ import annotations
from typing import Protocol, TYPE_CHECKING
from connectrpc.client import ConnectClient, ConnectClientSync
from connectrpc.code import Code
from connectrpc.compat import google_protobuf_binary_codec, google_protobuf_codecs
from connectrpc.compression.gzip import GzipCompression
from connectrpc.errors import ConnectError
from connectrpc.method import IdempotencyLevel, MethodInfo
from connectrpc.protocol import ProtocolType
from connectrpc.server import ConnectASGIApplication, ConnectWSGIApplication, Endpoint, EndpointSync
from pyqwest import Client, SyncClient
from .bot_live_pb2 import AcquireBotControlRequest, AcquireBotControlResponse, AttackEntityRequest, AttackEntityResponse, BotEvent, DigBlockRequest, DigBlockResponse, DismountRequest, DismountResponse, FindBlocksRequest, FindBlocksResponse, GetBlockRequest, GetBlockResponse, GoToRequest, InteractBlockRequest, InteractBlockResponse, InteractEntityRequest, InteractEntityResponse, ListNearbyEntitiesRequest, ListNearbyEntitiesResponse, MountEntityRequest, MountEntityResponse, PathfindProgress, PlaceBlockRequest, PlaceBlockResponse, ReleaseBotControlRequest, ReleaseBotControlResponse, ReleaseItemRequest, ReleaseItemResponse, RenewBotControlRequest, RenewBotControlResponse, RespawnRequest, RespawnResponse, RespondResourcePackRequest, RespondResourcePackResponse, SendChatRequest, SendChatResponse, SetCreativeSlotRequest, SetCreativeSlotResponse, SetFlyingRequest, SetFlyingResponse, SetVehicleControlRequest, SetVehicleControlResponse, SleepRequest, SleepResponse, StartElytraFlightRequest, StartElytraFlightResponse, StopPathfindingRequest, StopPathfindingResponse, SwingArmRequest, SwingArmResponse, UpdateSignRequest, UpdateSignResponse, UseItemRequest, UseItemResponse, WaitForChunksRequest, WaitForChunksResponse, WakeRequest, WakeResponse, WatchBotEventsRequest, WriteBookRequest, WriteBookResponse
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, AsyncIterator, Iterable, Iterator, Mapping
from connectrpc.codec import Codec
from connectrpc.compression import Compression
from connectrpc.interceptor import Interceptor, InterceptorSync
from connectrpc.request import Headers, RequestContext
_DEFAULT_CODECS = google_protobuf_codecs()
_PROTO_BINARY_CODEC = google_protobuf_binary_codec()
_GZIP_COMPRESSION = GzipCompression()
[docs]
class BotLiveService(Protocol):
[docs]
def watch_bot_events(self, request: WatchBotEventsRequest, ctx: RequestContext[WatchBotEventsRequest, BotEvent]) -> AsyncIterator[BotEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def send_chat(self, request: SendChatRequest, ctx: RequestContext[SendChatRequest, SendChatResponse]) -> SendChatResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_block(self, request: GetBlockRequest, ctx: RequestContext[GetBlockRequest, GetBlockResponse]) -> GetBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def find_blocks(self, request: FindBlocksRequest, ctx: RequestContext[FindBlocksRequest, FindBlocksResponse]) -> FindBlocksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def list_nearby_entities(self, request: ListNearbyEntitiesRequest, ctx: RequestContext[ListNearbyEntitiesRequest, ListNearbyEntitiesResponse]) -> ListNearbyEntitiesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def dig_block(self, request: DigBlockRequest, ctx: RequestContext[DigBlockRequest, DigBlockResponse]) -> DigBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def place_block(self, request: PlaceBlockRequest, ctx: RequestContext[PlaceBlockRequest, PlaceBlockResponse]) -> PlaceBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def interact_block(self, request: InteractBlockRequest, ctx: RequestContext[InteractBlockRequest, InteractBlockResponse]) -> InteractBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def use_item(self, request: UseItemRequest, ctx: RequestContext[UseItemRequest, UseItemResponse]) -> UseItemResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def release_item(self, request: ReleaseItemRequest, ctx: RequestContext[ReleaseItemRequest, ReleaseItemResponse]) -> ReleaseItemResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def attack_entity(self, request: AttackEntityRequest, ctx: RequestContext[AttackEntityRequest, AttackEntityResponse]) -> AttackEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def interact_entity(self, request: InteractEntityRequest, ctx: RequestContext[InteractEntityRequest, InteractEntityResponse]) -> InteractEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def swing_arm(self, request: SwingArmRequest, ctx: RequestContext[SwingArmRequest, SwingArmResponse]) -> SwingArmResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def respawn(self, request: RespawnRequest, ctx: RequestContext[RespawnRequest, RespawnResponse]) -> RespawnResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def sleep(self, request: SleepRequest, ctx: RequestContext[SleepRequest, SleepResponse]) -> SleepResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def wake(self, request: WakeRequest, ctx: RequestContext[WakeRequest, WakeResponse]) -> WakeResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def mount_entity(self, request: MountEntityRequest, ctx: RequestContext[MountEntityRequest, MountEntityResponse]) -> MountEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def dismount(self, request: DismountRequest, ctx: RequestContext[DismountRequest, DismountResponse]) -> DismountResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_vehicle_control(self, request: SetVehicleControlRequest, ctx: RequestContext[SetVehicleControlRequest, SetVehicleControlResponse]) -> SetVehicleControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def update_sign(self, request: UpdateSignRequest, ctx: RequestContext[UpdateSignRequest, UpdateSignResponse]) -> UpdateSignResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def write_book(self, request: WriteBookRequest, ctx: RequestContext[WriteBookRequest, WriteBookResponse]) -> WriteBookResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def respond_resource_pack(self, request: RespondResourcePackRequest, ctx: RequestContext[RespondResourcePackRequest, RespondResourcePackResponse]) -> RespondResourcePackResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_flying(self, request: SetFlyingRequest, ctx: RequestContext[SetFlyingRequest, SetFlyingResponse]) -> SetFlyingResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def start_elytra_flight(self, request: StartElytraFlightRequest, ctx: RequestContext[StartElytraFlightRequest, StartElytraFlightResponse]) -> StartElytraFlightResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_creative_slot(self, request: SetCreativeSlotRequest, ctx: RequestContext[SetCreativeSlotRequest, SetCreativeSlotResponse]) -> SetCreativeSlotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def wait_for_chunks(self, request: WaitForChunksRequest, ctx: RequestContext[WaitForChunksRequest, WaitForChunksResponse]) -> WaitForChunksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def go_to(self, request: GoToRequest, ctx: RequestContext[GoToRequest, PathfindProgress]) -> AsyncIterator[PathfindProgress]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def stop_pathfinding(self, request: StopPathfindingRequest, ctx: RequestContext[StopPathfindingRequest, StopPathfindingResponse]) -> StopPathfindingResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def acquire_bot_control(self, request: AcquireBotControlRequest, ctx: RequestContext[AcquireBotControlRequest, AcquireBotControlResponse]) -> AcquireBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def renew_bot_control(self, request: RenewBotControlRequest, ctx: RequestContext[RenewBotControlRequest, RenewBotControlResponse]) -> RenewBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def release_bot_control(self, request: ReleaseBotControlRequest, ctx: RequestContext[ReleaseBotControlRequest, ReleaseBotControlResponse]) -> ReleaseBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class BotLiveServiceASGIApplication(ConnectASGIApplication[BotLiveService]):
def __init__(
self,
service: BotLiveService | AsyncGenerator[BotLiveService],
*,
interceptors: Iterable[Interceptor] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
service=service,
endpoints=lambda svc: {
"/soulfire.v1.BotLiveService/WatchBotEvents": Endpoint.server_stream(
method=MethodInfo(
name="WatchBotEvents",
service_name="soulfire.v1.BotLiveService",
input=WatchBotEventsRequest,
output=BotEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.watch_bot_events,
),
"/soulfire.v1.BotLiveService/SendChat": Endpoint.unary(
method=MethodInfo(
name="SendChat",
service_name="soulfire.v1.BotLiveService",
input=SendChatRequest,
output=SendChatResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.send_chat,
),
"/soulfire.v1.BotLiveService/GetBlock": Endpoint.unary(
method=MethodInfo(
name="GetBlock",
service_name="soulfire.v1.BotLiveService",
input=GetBlockRequest,
output=GetBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_block,
),
"/soulfire.v1.BotLiveService/FindBlocks": Endpoint.unary(
method=MethodInfo(
name="FindBlocks",
service_name="soulfire.v1.BotLiveService",
input=FindBlocksRequest,
output=FindBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.find_blocks,
),
"/soulfire.v1.BotLiveService/ListNearbyEntities": Endpoint.unary(
method=MethodInfo(
name="ListNearbyEntities",
service_name="soulfire.v1.BotLiveService",
input=ListNearbyEntitiesRequest,
output=ListNearbyEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.list_nearby_entities,
),
"/soulfire.v1.BotLiveService/DigBlock": Endpoint.unary(
method=MethodInfo(
name="DigBlock",
service_name="soulfire.v1.BotLiveService",
input=DigBlockRequest,
output=DigBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.dig_block,
),
"/soulfire.v1.BotLiveService/PlaceBlock": Endpoint.unary(
method=MethodInfo(
name="PlaceBlock",
service_name="soulfire.v1.BotLiveService",
input=PlaceBlockRequest,
output=PlaceBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.place_block,
),
"/soulfire.v1.BotLiveService/InteractBlock": Endpoint.unary(
method=MethodInfo(
name="InteractBlock",
service_name="soulfire.v1.BotLiveService",
input=InteractBlockRequest,
output=InteractBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.interact_block,
),
"/soulfire.v1.BotLiveService/UseItem": Endpoint.unary(
method=MethodInfo(
name="UseItem",
service_name="soulfire.v1.BotLiveService",
input=UseItemRequest,
output=UseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.use_item,
),
"/soulfire.v1.BotLiveService/ReleaseItem": Endpoint.unary(
method=MethodInfo(
name="ReleaseItem",
service_name="soulfire.v1.BotLiveService",
input=ReleaseItemRequest,
output=ReleaseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.release_item,
),
"/soulfire.v1.BotLiveService/AttackEntity": Endpoint.unary(
method=MethodInfo(
name="AttackEntity",
service_name="soulfire.v1.BotLiveService",
input=AttackEntityRequest,
output=AttackEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.attack_entity,
),
"/soulfire.v1.BotLiveService/InteractEntity": Endpoint.unary(
method=MethodInfo(
name="InteractEntity",
service_name="soulfire.v1.BotLiveService",
input=InteractEntityRequest,
output=InteractEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.interact_entity,
),
"/soulfire.v1.BotLiveService/SwingArm": Endpoint.unary(
method=MethodInfo(
name="SwingArm",
service_name="soulfire.v1.BotLiveService",
input=SwingArmRequest,
output=SwingArmResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.swing_arm,
),
"/soulfire.v1.BotLiveService/Respawn": Endpoint.unary(
method=MethodInfo(
name="Respawn",
service_name="soulfire.v1.BotLiveService",
input=RespawnRequest,
output=RespawnResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.respawn,
),
"/soulfire.v1.BotLiveService/Sleep": Endpoint.unary(
method=MethodInfo(
name="Sleep",
service_name="soulfire.v1.BotLiveService",
input=SleepRequest,
output=SleepResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.sleep,
),
"/soulfire.v1.BotLiveService/Wake": Endpoint.unary(
method=MethodInfo(
name="Wake",
service_name="soulfire.v1.BotLiveService",
input=WakeRequest,
output=WakeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.wake,
),
"/soulfire.v1.BotLiveService/MountEntity": Endpoint.unary(
method=MethodInfo(
name="MountEntity",
service_name="soulfire.v1.BotLiveService",
input=MountEntityRequest,
output=MountEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.mount_entity,
),
"/soulfire.v1.BotLiveService/Dismount": Endpoint.unary(
method=MethodInfo(
name="Dismount",
service_name="soulfire.v1.BotLiveService",
input=DismountRequest,
output=DismountResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.dismount,
),
"/soulfire.v1.BotLiveService/SetVehicleControl": Endpoint.unary(
method=MethodInfo(
name="SetVehicleControl",
service_name="soulfire.v1.BotLiveService",
input=SetVehicleControlRequest,
output=SetVehicleControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_vehicle_control,
),
"/soulfire.v1.BotLiveService/UpdateSign": Endpoint.unary(
method=MethodInfo(
name="UpdateSign",
service_name="soulfire.v1.BotLiveService",
input=UpdateSignRequest,
output=UpdateSignResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.update_sign,
),
"/soulfire.v1.BotLiveService/WriteBook": Endpoint.unary(
method=MethodInfo(
name="WriteBook",
service_name="soulfire.v1.BotLiveService",
input=WriteBookRequest,
output=WriteBookResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.write_book,
),
"/soulfire.v1.BotLiveService/RespondResourcePack": Endpoint.unary(
method=MethodInfo(
name="RespondResourcePack",
service_name="soulfire.v1.BotLiveService",
input=RespondResourcePackRequest,
output=RespondResourcePackResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.respond_resource_pack,
),
"/soulfire.v1.BotLiveService/SetFlying": Endpoint.unary(
method=MethodInfo(
name="SetFlying",
service_name="soulfire.v1.BotLiveService",
input=SetFlyingRequest,
output=SetFlyingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_flying,
),
"/soulfire.v1.BotLiveService/StartElytraFlight": Endpoint.unary(
method=MethodInfo(
name="StartElytraFlight",
service_name="soulfire.v1.BotLiveService",
input=StartElytraFlightRequest,
output=StartElytraFlightResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.start_elytra_flight,
),
"/soulfire.v1.BotLiveService/SetCreativeSlot": Endpoint.unary(
method=MethodInfo(
name="SetCreativeSlot",
service_name="soulfire.v1.BotLiveService",
input=SetCreativeSlotRequest,
output=SetCreativeSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_creative_slot,
),
"/soulfire.v1.BotLiveService/WaitForChunks": Endpoint.unary(
method=MethodInfo(
name="WaitForChunks",
service_name="soulfire.v1.BotLiveService",
input=WaitForChunksRequest,
output=WaitForChunksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.wait_for_chunks,
),
"/soulfire.v1.BotLiveService/GoTo": Endpoint.server_stream(
method=MethodInfo(
name="GoTo",
service_name="soulfire.v1.BotLiveService",
input=GoToRequest,
output=PathfindProgress,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.go_to,
),
"/soulfire.v1.BotLiveService/StopPathfinding": Endpoint.unary(
method=MethodInfo(
name="StopPathfinding",
service_name="soulfire.v1.BotLiveService",
input=StopPathfindingRequest,
output=StopPathfindingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.stop_pathfinding,
),
"/soulfire.v1.BotLiveService/AcquireBotControl": Endpoint.unary(
method=MethodInfo(
name="AcquireBotControl",
service_name="soulfire.v1.BotLiveService",
input=AcquireBotControlRequest,
output=AcquireBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.acquire_bot_control,
),
"/soulfire.v1.BotLiveService/RenewBotControl": Endpoint.unary(
method=MethodInfo(
name="RenewBotControl",
service_name="soulfire.v1.BotLiveService",
input=RenewBotControlRequest,
output=RenewBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.renew_bot_control,
),
"/soulfire.v1.BotLiveService/ReleaseBotControl": Endpoint.unary(
method=MethodInfo(
name="ReleaseBotControl",
service_name="soulfire.v1.BotLiveService",
input=ReleaseBotControlRequest,
output=ReleaseBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.release_bot_control,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.BotLiveService"
[docs]
class BotLiveServiceClient(ConnectClient):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[Interceptor] = (),
http_client: Client | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
def watch_bot_events(
self,
request: WatchBotEventsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[BotEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotEvents",
service_name="soulfire.v1.BotLiveService",
input=WatchBotEventsRequest,
output=BotEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def send_chat(
self,
request: SendChatRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SendChatResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SendChat",
service_name="soulfire.v1.BotLiveService",
input=SendChatRequest,
output=SendChatResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_block(
self,
request: GetBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetBlock",
service_name="soulfire.v1.BotLiveService",
input=GetBlockRequest,
output=GetBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def find_blocks(
self,
request: FindBlocksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> FindBlocksResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="FindBlocks",
service_name="soulfire.v1.BotLiveService",
input=FindBlocksRequest,
output=FindBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def list_nearby_entities(
self,
request: ListNearbyEntitiesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListNearbyEntitiesResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ListNearbyEntities",
service_name="soulfire.v1.BotLiveService",
input=ListNearbyEntitiesRequest,
output=ListNearbyEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def dig_block(
self,
request: DigBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DigBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="DigBlock",
service_name="soulfire.v1.BotLiveService",
input=DigBlockRequest,
output=DigBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def place_block(
self,
request: PlaceBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> PlaceBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="PlaceBlock",
service_name="soulfire.v1.BotLiveService",
input=PlaceBlockRequest,
output=PlaceBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def interact_block(
self,
request: InteractBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InteractBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="InteractBlock",
service_name="soulfire.v1.BotLiveService",
input=InteractBlockRequest,
output=InteractBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def use_item(
self,
request: UseItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UseItemResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UseItem",
service_name="soulfire.v1.BotLiveService",
input=UseItemRequest,
output=UseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def release_item(
self,
request: ReleaseItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ReleaseItemResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ReleaseItem",
service_name="soulfire.v1.BotLiveService",
input=ReleaseItemRequest,
output=ReleaseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def attack_entity(
self,
request: AttackEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AttackEntityResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="AttackEntity",
service_name="soulfire.v1.BotLiveService",
input=AttackEntityRequest,
output=AttackEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def interact_entity(
self,
request: InteractEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InteractEntityResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="InteractEntity",
service_name="soulfire.v1.BotLiveService",
input=InteractEntityRequest,
output=InteractEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def swing_arm(
self,
request: SwingArmRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SwingArmResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SwingArm",
service_name="soulfire.v1.BotLiveService",
input=SwingArmRequest,
output=SwingArmResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def respawn(
self,
request: RespawnRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RespawnResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="Respawn",
service_name="soulfire.v1.BotLiveService",
input=RespawnRequest,
output=RespawnResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def sleep(
self,
request: SleepRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SleepResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="Sleep",
service_name="soulfire.v1.BotLiveService",
input=SleepRequest,
output=SleepResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def wake(
self,
request: WakeRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WakeResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="Wake",
service_name="soulfire.v1.BotLiveService",
input=WakeRequest,
output=WakeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def mount_entity(
self,
request: MountEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> MountEntityResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="MountEntity",
service_name="soulfire.v1.BotLiveService",
input=MountEntityRequest,
output=MountEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def dismount(
self,
request: DismountRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DismountResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="Dismount",
service_name="soulfire.v1.BotLiveService",
input=DismountRequest,
output=DismountResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_vehicle_control(
self,
request: SetVehicleControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetVehicleControlResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetVehicleControl",
service_name="soulfire.v1.BotLiveService",
input=SetVehicleControlRequest,
output=SetVehicleControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def update_sign(
self,
request: UpdateSignRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSignResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSign",
service_name="soulfire.v1.BotLiveService",
input=UpdateSignRequest,
output=UpdateSignResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def write_book(
self,
request: WriteBookRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WriteBookResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="WriteBook",
service_name="soulfire.v1.BotLiveService",
input=WriteBookRequest,
output=WriteBookResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def respond_resource_pack(
self,
request: RespondResourcePackRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RespondResourcePackResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="RespondResourcePack",
service_name="soulfire.v1.BotLiveService",
input=RespondResourcePackRequest,
output=RespondResourcePackResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_flying(
self,
request: SetFlyingRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetFlyingResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetFlying",
service_name="soulfire.v1.BotLiveService",
input=SetFlyingRequest,
output=SetFlyingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def start_elytra_flight(
self,
request: StartElytraFlightRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> StartElytraFlightResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="StartElytraFlight",
service_name="soulfire.v1.BotLiveService",
input=StartElytraFlightRequest,
output=StartElytraFlightResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_creative_slot(
self,
request: SetCreativeSlotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetCreativeSlotResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetCreativeSlot",
service_name="soulfire.v1.BotLiveService",
input=SetCreativeSlotRequest,
output=SetCreativeSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def wait_for_chunks(
self,
request: WaitForChunksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WaitForChunksResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="WaitForChunks",
service_name="soulfire.v1.BotLiveService",
input=WaitForChunksRequest,
output=WaitForChunksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def go_to(
self,
request: GoToRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[PathfindProgress]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="GoTo",
service_name="soulfire.v1.BotLiveService",
input=GoToRequest,
output=PathfindProgress,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def stop_pathfinding(
self,
request: StopPathfindingRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> StopPathfindingResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="StopPathfinding",
service_name="soulfire.v1.BotLiveService",
input=StopPathfindingRequest,
output=StopPathfindingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def acquire_bot_control(
self,
request: AcquireBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AcquireBotControlResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="AcquireBotControl",
service_name="soulfire.v1.BotLiveService",
input=AcquireBotControlRequest,
output=AcquireBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def renew_bot_control(
self,
request: RenewBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RenewBotControlResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="RenewBotControl",
service_name="soulfire.v1.BotLiveService",
input=RenewBotControlRequest,
output=RenewBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def release_bot_control(
self,
request: ReleaseBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ReleaseBotControlResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ReleaseBotControl",
service_name="soulfire.v1.BotLiveService",
input=ReleaseBotControlRequest,
output=ReleaseBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class BotLiveServiceSync(Protocol):
[docs]
def watch_bot_events(self, request: WatchBotEventsRequest, ctx: RequestContext[WatchBotEventsRequest, BotEvent]) -> Iterator[BotEvent]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def send_chat(self, request: SendChatRequest, ctx: RequestContext[SendChatRequest, SendChatResponse]) -> SendChatResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_block(self, request: GetBlockRequest, ctx: RequestContext[GetBlockRequest, GetBlockResponse]) -> GetBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def find_blocks(self, request: FindBlocksRequest, ctx: RequestContext[FindBlocksRequest, FindBlocksResponse]) -> FindBlocksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def list_nearby_entities(self, request: ListNearbyEntitiesRequest, ctx: RequestContext[ListNearbyEntitiesRequest, ListNearbyEntitiesResponse]) -> ListNearbyEntitiesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def dig_block(self, request: DigBlockRequest, ctx: RequestContext[DigBlockRequest, DigBlockResponse]) -> DigBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def place_block(self, request: PlaceBlockRequest, ctx: RequestContext[PlaceBlockRequest, PlaceBlockResponse]) -> PlaceBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def interact_block(self, request: InteractBlockRequest, ctx: RequestContext[InteractBlockRequest, InteractBlockResponse]) -> InteractBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def use_item(self, request: UseItemRequest, ctx: RequestContext[UseItemRequest, UseItemResponse]) -> UseItemResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def release_item(self, request: ReleaseItemRequest, ctx: RequestContext[ReleaseItemRequest, ReleaseItemResponse]) -> ReleaseItemResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def attack_entity(self, request: AttackEntityRequest, ctx: RequestContext[AttackEntityRequest, AttackEntityResponse]) -> AttackEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def interact_entity(self, request: InteractEntityRequest, ctx: RequestContext[InteractEntityRequest, InteractEntityResponse]) -> InteractEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def swing_arm(self, request: SwingArmRequest, ctx: RequestContext[SwingArmRequest, SwingArmResponse]) -> SwingArmResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def respawn(self, request: RespawnRequest, ctx: RequestContext[RespawnRequest, RespawnResponse]) -> RespawnResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def sleep(self, request: SleepRequest, ctx: RequestContext[SleepRequest, SleepResponse]) -> SleepResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def wake(self, request: WakeRequest, ctx: RequestContext[WakeRequest, WakeResponse]) -> WakeResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def mount_entity(self, request: MountEntityRequest, ctx: RequestContext[MountEntityRequest, MountEntityResponse]) -> MountEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def dismount(self, request: DismountRequest, ctx: RequestContext[DismountRequest, DismountResponse]) -> DismountResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_vehicle_control(self, request: SetVehicleControlRequest, ctx: RequestContext[SetVehicleControlRequest, SetVehicleControlResponse]) -> SetVehicleControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def update_sign(self, request: UpdateSignRequest, ctx: RequestContext[UpdateSignRequest, UpdateSignResponse]) -> UpdateSignResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def write_book(self, request: WriteBookRequest, ctx: RequestContext[WriteBookRequest, WriteBookResponse]) -> WriteBookResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def respond_resource_pack(self, request: RespondResourcePackRequest, ctx: RequestContext[RespondResourcePackRequest, RespondResourcePackResponse]) -> RespondResourcePackResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_flying(self, request: SetFlyingRequest, ctx: RequestContext[SetFlyingRequest, SetFlyingResponse]) -> SetFlyingResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def start_elytra_flight(self, request: StartElytraFlightRequest, ctx: RequestContext[StartElytraFlightRequest, StartElytraFlightResponse]) -> StartElytraFlightResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_creative_slot(self, request: SetCreativeSlotRequest, ctx: RequestContext[SetCreativeSlotRequest, SetCreativeSlotResponse]) -> SetCreativeSlotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def wait_for_chunks(self, request: WaitForChunksRequest, ctx: RequestContext[WaitForChunksRequest, WaitForChunksResponse]) -> WaitForChunksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def go_to(self, request: GoToRequest, ctx: RequestContext[GoToRequest, PathfindProgress]) -> Iterator[PathfindProgress]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def stop_pathfinding(self, request: StopPathfindingRequest, ctx: RequestContext[StopPathfindingRequest, StopPathfindingResponse]) -> StopPathfindingResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def acquire_bot_control(self, request: AcquireBotControlRequest, ctx: RequestContext[AcquireBotControlRequest, AcquireBotControlResponse]) -> AcquireBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def renew_bot_control(self, request: RenewBotControlRequest, ctx: RequestContext[RenewBotControlRequest, RenewBotControlResponse]) -> RenewBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def release_bot_control(self, request: ReleaseBotControlRequest, ctx: RequestContext[ReleaseBotControlRequest, ReleaseBotControlResponse]) -> ReleaseBotControlResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class BotLiveServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: BotLiveServiceSync,
interceptors: Iterable[InterceptorSync] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
endpoints={
"/soulfire.v1.BotLiveService/WatchBotEvents": EndpointSync.server_stream(
method=MethodInfo(
name="WatchBotEvents",
service_name="soulfire.v1.BotLiveService",
input=WatchBotEventsRequest,
output=BotEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.watch_bot_events,
),
"/soulfire.v1.BotLiveService/SendChat": EndpointSync.unary(
method=MethodInfo(
name="SendChat",
service_name="soulfire.v1.BotLiveService",
input=SendChatRequest,
output=SendChatResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.send_chat,
),
"/soulfire.v1.BotLiveService/GetBlock": EndpointSync.unary(
method=MethodInfo(
name="GetBlock",
service_name="soulfire.v1.BotLiveService",
input=GetBlockRequest,
output=GetBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_block,
),
"/soulfire.v1.BotLiveService/FindBlocks": EndpointSync.unary(
method=MethodInfo(
name="FindBlocks",
service_name="soulfire.v1.BotLiveService",
input=FindBlocksRequest,
output=FindBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.find_blocks,
),
"/soulfire.v1.BotLiveService/ListNearbyEntities": EndpointSync.unary(
method=MethodInfo(
name="ListNearbyEntities",
service_name="soulfire.v1.BotLiveService",
input=ListNearbyEntitiesRequest,
output=ListNearbyEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.list_nearby_entities,
),
"/soulfire.v1.BotLiveService/DigBlock": EndpointSync.unary(
method=MethodInfo(
name="DigBlock",
service_name="soulfire.v1.BotLiveService",
input=DigBlockRequest,
output=DigBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.dig_block,
),
"/soulfire.v1.BotLiveService/PlaceBlock": EndpointSync.unary(
method=MethodInfo(
name="PlaceBlock",
service_name="soulfire.v1.BotLiveService",
input=PlaceBlockRequest,
output=PlaceBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.place_block,
),
"/soulfire.v1.BotLiveService/InteractBlock": EndpointSync.unary(
method=MethodInfo(
name="InteractBlock",
service_name="soulfire.v1.BotLiveService",
input=InteractBlockRequest,
output=InteractBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.interact_block,
),
"/soulfire.v1.BotLiveService/UseItem": EndpointSync.unary(
method=MethodInfo(
name="UseItem",
service_name="soulfire.v1.BotLiveService",
input=UseItemRequest,
output=UseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.use_item,
),
"/soulfire.v1.BotLiveService/ReleaseItem": EndpointSync.unary(
method=MethodInfo(
name="ReleaseItem",
service_name="soulfire.v1.BotLiveService",
input=ReleaseItemRequest,
output=ReleaseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.release_item,
),
"/soulfire.v1.BotLiveService/AttackEntity": EndpointSync.unary(
method=MethodInfo(
name="AttackEntity",
service_name="soulfire.v1.BotLiveService",
input=AttackEntityRequest,
output=AttackEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.attack_entity,
),
"/soulfire.v1.BotLiveService/InteractEntity": EndpointSync.unary(
method=MethodInfo(
name="InteractEntity",
service_name="soulfire.v1.BotLiveService",
input=InteractEntityRequest,
output=InteractEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.interact_entity,
),
"/soulfire.v1.BotLiveService/SwingArm": EndpointSync.unary(
method=MethodInfo(
name="SwingArm",
service_name="soulfire.v1.BotLiveService",
input=SwingArmRequest,
output=SwingArmResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.swing_arm,
),
"/soulfire.v1.BotLiveService/Respawn": EndpointSync.unary(
method=MethodInfo(
name="Respawn",
service_name="soulfire.v1.BotLiveService",
input=RespawnRequest,
output=RespawnResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.respawn,
),
"/soulfire.v1.BotLiveService/Sleep": EndpointSync.unary(
method=MethodInfo(
name="Sleep",
service_name="soulfire.v1.BotLiveService",
input=SleepRequest,
output=SleepResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.sleep,
),
"/soulfire.v1.BotLiveService/Wake": EndpointSync.unary(
method=MethodInfo(
name="Wake",
service_name="soulfire.v1.BotLiveService",
input=WakeRequest,
output=WakeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.wake,
),
"/soulfire.v1.BotLiveService/MountEntity": EndpointSync.unary(
method=MethodInfo(
name="MountEntity",
service_name="soulfire.v1.BotLiveService",
input=MountEntityRequest,
output=MountEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.mount_entity,
),
"/soulfire.v1.BotLiveService/Dismount": EndpointSync.unary(
method=MethodInfo(
name="Dismount",
service_name="soulfire.v1.BotLiveService",
input=DismountRequest,
output=DismountResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.dismount,
),
"/soulfire.v1.BotLiveService/SetVehicleControl": EndpointSync.unary(
method=MethodInfo(
name="SetVehicleControl",
service_name="soulfire.v1.BotLiveService",
input=SetVehicleControlRequest,
output=SetVehicleControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_vehicle_control,
),
"/soulfire.v1.BotLiveService/UpdateSign": EndpointSync.unary(
method=MethodInfo(
name="UpdateSign",
service_name="soulfire.v1.BotLiveService",
input=UpdateSignRequest,
output=UpdateSignResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.update_sign,
),
"/soulfire.v1.BotLiveService/WriteBook": EndpointSync.unary(
method=MethodInfo(
name="WriteBook",
service_name="soulfire.v1.BotLiveService",
input=WriteBookRequest,
output=WriteBookResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.write_book,
),
"/soulfire.v1.BotLiveService/RespondResourcePack": EndpointSync.unary(
method=MethodInfo(
name="RespondResourcePack",
service_name="soulfire.v1.BotLiveService",
input=RespondResourcePackRequest,
output=RespondResourcePackResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.respond_resource_pack,
),
"/soulfire.v1.BotLiveService/SetFlying": EndpointSync.unary(
method=MethodInfo(
name="SetFlying",
service_name="soulfire.v1.BotLiveService",
input=SetFlyingRequest,
output=SetFlyingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_flying,
),
"/soulfire.v1.BotLiveService/StartElytraFlight": EndpointSync.unary(
method=MethodInfo(
name="StartElytraFlight",
service_name="soulfire.v1.BotLiveService",
input=StartElytraFlightRequest,
output=StartElytraFlightResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.start_elytra_flight,
),
"/soulfire.v1.BotLiveService/SetCreativeSlot": EndpointSync.unary(
method=MethodInfo(
name="SetCreativeSlot",
service_name="soulfire.v1.BotLiveService",
input=SetCreativeSlotRequest,
output=SetCreativeSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_creative_slot,
),
"/soulfire.v1.BotLiveService/WaitForChunks": EndpointSync.unary(
method=MethodInfo(
name="WaitForChunks",
service_name="soulfire.v1.BotLiveService",
input=WaitForChunksRequest,
output=WaitForChunksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.wait_for_chunks,
),
"/soulfire.v1.BotLiveService/GoTo": EndpointSync.server_stream(
method=MethodInfo(
name="GoTo",
service_name="soulfire.v1.BotLiveService",
input=GoToRequest,
output=PathfindProgress,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.go_to,
),
"/soulfire.v1.BotLiveService/StopPathfinding": EndpointSync.unary(
method=MethodInfo(
name="StopPathfinding",
service_name="soulfire.v1.BotLiveService",
input=StopPathfindingRequest,
output=StopPathfindingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.stop_pathfinding,
),
"/soulfire.v1.BotLiveService/AcquireBotControl": EndpointSync.unary(
method=MethodInfo(
name="AcquireBotControl",
service_name="soulfire.v1.BotLiveService",
input=AcquireBotControlRequest,
output=AcquireBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.acquire_bot_control,
),
"/soulfire.v1.BotLiveService/RenewBotControl": EndpointSync.unary(
method=MethodInfo(
name="RenewBotControl",
service_name="soulfire.v1.BotLiveService",
input=RenewBotControlRequest,
output=RenewBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.renew_bot_control,
),
"/soulfire.v1.BotLiveService/ReleaseBotControl": EndpointSync.unary(
method=MethodInfo(
name="ReleaseBotControl",
service_name="soulfire.v1.BotLiveService",
input=ReleaseBotControlRequest,
output=ReleaseBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.release_bot_control,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.BotLiveService"
[docs]
class BotLiveServiceClientSync(ConnectClientSync):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[InterceptorSync] = (),
http_client: SyncClient | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
def watch_bot_events(
self,
request: WatchBotEventsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[BotEvent]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotEvents",
service_name="soulfire.v1.BotLiveService",
input=WatchBotEventsRequest,
output=BotEvent,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def send_chat(
self,
request: SendChatRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SendChatResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SendChat",
service_name="soulfire.v1.BotLiveService",
input=SendChatRequest,
output=SendChatResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_block(
self,
request: GetBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetBlock",
service_name="soulfire.v1.BotLiveService",
input=GetBlockRequest,
output=GetBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def find_blocks(
self,
request: FindBlocksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> FindBlocksResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="FindBlocks",
service_name="soulfire.v1.BotLiveService",
input=FindBlocksRequest,
output=FindBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def list_nearby_entities(
self,
request: ListNearbyEntitiesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ListNearbyEntitiesResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ListNearbyEntities",
service_name="soulfire.v1.BotLiveService",
input=ListNearbyEntitiesRequest,
output=ListNearbyEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def dig_block(
self,
request: DigBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DigBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="DigBlock",
service_name="soulfire.v1.BotLiveService",
input=DigBlockRequest,
output=DigBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def place_block(
self,
request: PlaceBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> PlaceBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="PlaceBlock",
service_name="soulfire.v1.BotLiveService",
input=PlaceBlockRequest,
output=PlaceBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def interact_block(
self,
request: InteractBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InteractBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="InteractBlock",
service_name="soulfire.v1.BotLiveService",
input=InteractBlockRequest,
output=InteractBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def use_item(
self,
request: UseItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UseItemResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UseItem",
service_name="soulfire.v1.BotLiveService",
input=UseItemRequest,
output=UseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def release_item(
self,
request: ReleaseItemRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ReleaseItemResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ReleaseItem",
service_name="soulfire.v1.BotLiveService",
input=ReleaseItemRequest,
output=ReleaseItemResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def attack_entity(
self,
request: AttackEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AttackEntityResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="AttackEntity",
service_name="soulfire.v1.BotLiveService",
input=AttackEntityRequest,
output=AttackEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def interact_entity(
self,
request: InteractEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> InteractEntityResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="InteractEntity",
service_name="soulfire.v1.BotLiveService",
input=InteractEntityRequest,
output=InteractEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def swing_arm(
self,
request: SwingArmRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SwingArmResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SwingArm",
service_name="soulfire.v1.BotLiveService",
input=SwingArmRequest,
output=SwingArmResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def respawn(
self,
request: RespawnRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RespawnResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="Respawn",
service_name="soulfire.v1.BotLiveService",
input=RespawnRequest,
output=RespawnResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def sleep(
self,
request: SleepRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SleepResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="Sleep",
service_name="soulfire.v1.BotLiveService",
input=SleepRequest,
output=SleepResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def wake(
self,
request: WakeRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WakeResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="Wake",
service_name="soulfire.v1.BotLiveService",
input=WakeRequest,
output=WakeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def mount_entity(
self,
request: MountEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> MountEntityResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="MountEntity",
service_name="soulfire.v1.BotLiveService",
input=MountEntityRequest,
output=MountEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def dismount(
self,
request: DismountRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> DismountResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="Dismount",
service_name="soulfire.v1.BotLiveService",
input=DismountRequest,
output=DismountResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_vehicle_control(
self,
request: SetVehicleControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetVehicleControlResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetVehicleControl",
service_name="soulfire.v1.BotLiveService",
input=SetVehicleControlRequest,
output=SetVehicleControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def update_sign(
self,
request: UpdateSignRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> UpdateSignResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateSign",
service_name="soulfire.v1.BotLiveService",
input=UpdateSignRequest,
output=UpdateSignResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def write_book(
self,
request: WriteBookRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WriteBookResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="WriteBook",
service_name="soulfire.v1.BotLiveService",
input=WriteBookRequest,
output=WriteBookResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def respond_resource_pack(
self,
request: RespondResourcePackRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RespondResourcePackResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="RespondResourcePack",
service_name="soulfire.v1.BotLiveService",
input=RespondResourcePackRequest,
output=RespondResourcePackResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_flying(
self,
request: SetFlyingRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetFlyingResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetFlying",
service_name="soulfire.v1.BotLiveService",
input=SetFlyingRequest,
output=SetFlyingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def start_elytra_flight(
self,
request: StartElytraFlightRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> StartElytraFlightResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="StartElytraFlight",
service_name="soulfire.v1.BotLiveService",
input=StartElytraFlightRequest,
output=StartElytraFlightResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_creative_slot(
self,
request: SetCreativeSlotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetCreativeSlotResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetCreativeSlot",
service_name="soulfire.v1.BotLiveService",
input=SetCreativeSlotRequest,
output=SetCreativeSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def wait_for_chunks(
self,
request: WaitForChunksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> WaitForChunksResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="WaitForChunks",
service_name="soulfire.v1.BotLiveService",
input=WaitForChunksRequest,
output=WaitForChunksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def go_to(
self,
request: GoToRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[PathfindProgress]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="GoTo",
service_name="soulfire.v1.BotLiveService",
input=GoToRequest,
output=PathfindProgress,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def stop_pathfinding(
self,
request: StopPathfindingRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> StopPathfindingResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="StopPathfinding",
service_name="soulfire.v1.BotLiveService",
input=StopPathfindingRequest,
output=StopPathfindingResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def acquire_bot_control(
self,
request: AcquireBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AcquireBotControlResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="AcquireBotControl",
service_name="soulfire.v1.BotLiveService",
input=AcquireBotControlRequest,
output=AcquireBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def renew_bot_control(
self,
request: RenewBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RenewBotControlResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="RenewBotControl",
service_name="soulfire.v1.BotLiveService",
input=RenewBotControlRequest,
output=RenewBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def release_bot_control(
self,
request: ReleaseBotControlRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> ReleaseBotControlResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ReleaseBotControl",
service_name="soulfire.v1.BotLiveService",
input=ReleaseBotControlRequest,
output=ReleaseBotControlResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)