Shared events and plugin RPC

The shared event stream behind client.events, and the typed plugin RPC helpers behind client.rpc(Definition). See the events guide and the plugin RPC guide.

Event hubs

EventHub

EventHub(
    client: Opencode, *, reconnect: bool | Reconnect = False
)

One shared event connection for any number of synchronous subscribers.

client.events is the hub of a client. Create another hub with different options, such as reconnection, with EventHub(client, reconnect=True). A background thread reads the connection while it is open.

Example::

with client.events.of_type(SessionIdleEvent).for_session(session.id) as idle:
    client.session.prompt(session.id, text="Hello")
    next(idle)

Parameters:

Name	Type	Description	Default
client	Opencode	The client that opens the connection.	required
reconnect	bool | Reconnect	True or a :class:Reconnect policy to reopen the connection after it drops. Off by default.	False

subscribe

subscribe(
    *, on_activity: Callable[[], object] | None = None
) -> Subscription[Event]

Returns a subscription to every event. It joins the shared connection when first iterated.

Parameters:

Name	Type	Description	Default
on_activity	Callable[[], object] | None	Called on the reader thread for every chunk received, including keepalive comments. Use it as a liveness signal; keep it fast.	None

of_type

of_type(*types: type[_T]) -> Subscription[_T]

Shortcut for subscribe().of_type(*types).

for_session

for_session(session_id: str) -> Subscription[Event]

Shortcut for subscribe().for_session(session_id).

AsyncEventHub

AsyncEventHub(
    client: AsyncOpencode,
    *,
    reconnect: bool | Reconnect = False,
)

One shared event connection for any number of asynchronous subscribers. Works with asyncio and trio.

client.events is the hub of a client. Create another hub with different options, such as reconnection, with AsyncEventHub(client, reconnect=True). The connection is read by a detached task of the running event loop, and a hub must only be used from one event loop.

Example::

async with client.events.of_type(SessionIdleEvent).for_session(session.id) as idle:
    await client.session.prompt(session.id, text="Hello")
    await idle.get()

Parameters:

Name	Type	Description	Default
client	AsyncOpencode	The client that opens the connection.	required
reconnect	bool | Reconnect	True or a :class:Reconnect policy to reopen the connection after it drops. Off by default.	False

subscribe

subscribe(
    *, on_activity: Callable[[], object] | None = None
) -> AsyncSubscription[Event]

Returns a subscription to every event. It joins the shared connection when first iterated.

Parameters:

Name	Type	Description	Default
on_activity	Callable[[], object] | None	Called for every chunk received, including keepalive comments. Use it as a liveness signal; keep it fast and non-blocking.	None

of_type

of_type(*types: type[_T]) -> AsyncSubscription[_T]

Shortcut for subscribe().of_type(*types).

for_session

for_session(session_id: str) -> AsyncSubscription[Event]

Shortcut for subscribe().for_session(session_id).

Subscription

Subscription(
    hub: EventHub,
    predicates: tuple[Callable[[Any], bool], ...],
    on_activity: Callable[[], object] | None,
)

Bases: _Filters[_E]

One consumer of a shared event stream. Iterate it with for; close it, or use with, to leave.

The subscription joins the shared connection when entered as a context manager or first iterated, so filters can be chained before that::

for event in client.events.for_session(session_id).of_type(SessionIdleEvent):
    ...

Iteration ends when the connection ends, and raises the connection's error if it failed, after delivering the events already queued. close() may be called from any thread to stop a blocked iteration.

of_type

of_type(*types: type[_T]) -> Subscription[_T]

Returns a subscription that only yields events that are instances of types.

Type checkers narrow the yielded type, so of_type(SessionIdleEvent) yields SessionIdleEvent.

for_session

for_session(session_id: str) -> Subscription[_E]

Returns a subscription that only yields events of one session (see :func:session_id_of).

where

where(
    predicate: Callable[[_E], bool],
) -> Subscription[_E]

Returns a subscription that only yields events for which predicate returns true.

The predicate runs on the reader thread, after the other filters of this subscription.

get

get(timeout: float | None = None) -> _E

Returns the next event, waiting up to timeout seconds (forever when None).

Raises:

Type	Description
StopIteration	The connection ended or the subscription was closed.
TimeoutError	No event arrived in time. The subscription stays usable.
EventOverflowError	This subscriber fell too far behind.

close

close() -> None

Leaves the shared stream. Pending and future iterations end. Safe to call more than once.

AsyncSubscription

AsyncSubscription(
    hub: AsyncEventHub,
    predicates: tuple[Callable[[Any], bool], ...],
    on_activity: Callable[[], object] | None,
)

Bases: _Filters[_E]

One consumer of a shared event stream. Iterate it with async for; close it, or use async with.

The subscription joins the shared connection when entered as a context manager or first iterated, so filters can be chained before that::

async for event in client.events.for_session(session_id).of_type(SessionIdleEvent):
    ...

Iteration ends when the connection ends, and raises the connection's error if it failed, after delivering the events already queued. Cancelling a task that waits for an event (anyio.fail_after, asyncio.timeout, trio cancel scopes) leaves the subscription usable; close it to leave the stream.

of_type

of_type(*types: type[_T]) -> AsyncSubscription[_T]

Returns a subscription that only yields events that are instances of types.

Type checkers narrow the yielded type, so of_type(SessionIdleEvent) yields SessionIdleEvent.

for_session

for_session(session_id: str) -> AsyncSubscription[_E]

Returns a subscription that only yields events of one session (see :func:session_id_of).

where

where(
    predicate: Callable[[_E], bool],
) -> AsyncSubscription[_E]

Returns a subscription that only yields events for which predicate returns true.

get async

get() -> _E

Returns the next event. Bound the wait with anyio.fail_after or your event loop's timeout.

Raises:

Type	Description
EndOfStream	The connection ended or the subscription was closed.
EventOverflowError	This subscriber fell too far behind.

close async

close() -> None

Leaves the shared stream. Pending and future iterations end. Safe to call more than once.

close_nowait

close_nowait() -> None

Synchronous form of :meth:close, for callbacks and finally blocks that cannot await.

Reconnect dataclass

Reconnect(
    initial_delay: float = 0.5,
    max_delay: float = 30.0,
    max_attempts: int | None = None,
)

Opt-in reconnection policy for an event hub.

After the connection ends or fails with a connection error, a 5xx or a 429, the hub waits and connects again. The delay starts at initial_delay seconds, doubles per failed attempt up to max_delay, gets up to 25% jitter, and resets once a connection delivers an event. Other errors (authentication, malformed events) still end every subscription.

Events published while disconnected are lost. The server sends server.connected again on every connection; subscribers see it only once, and late subscribers get the latest one.

Parameters:

Name	Type	Description	Default
initial_delay	float	Seconds before the first reconnection attempt.	0.5
max_delay	float	Upper bound for the delay between attempts.	30.0
max_attempts	int | None	Consecutive failed attempts before giving up, or None to retry forever.	None

initial_delay class-attribute instance-attribute

initial_delay: float = 0.5

max_delay class-attribute instance-attribute

max_delay: float = 30.0

max_attempts class-attribute instance-attribute

max_attempts: int | None = None

delay

delay(attempt: int) -> float

Seconds to wait before reconnection attempt attempt (counting from zero).

allows

allows(attempt: int) -> bool

Whether reconnection attempt attempt (counting from zero) may run.

EventOverflowError

Bases: OpencodeError

A subscriber fell more than :data:SUBSCRIBER_CAPACITY events behind the shared stream.

Only that subscriber fails. Its queued events are discarded, since they no longer form a gap-free sequence.

SUBSCRIBER_CAPACITY module-attribute

SUBSCRIBER_CAPACITY = 4096

Events a subscriber may have queued before it fails with :class:EventOverflowError. Matches the TS client.

session_id_of

session_id_of(event: object) -> str | None

Returns the session an event belongs to, or None for events outside any session.

Looks for data.session_id on typed events and data["sessionID"] on raw ones.

Plugin RPC

TypedRpcResource

TypedRpcResource(client: Opencode)

Bases: RpcResource

client.rpc: the generated call plus client.rpc(Definition) for typed access.

RpcDefinition

Base class for plugin RPC declarations. Pass the RPC id as a class keyword::

class Counter(RpcDefinition, id="counter"):
    add = rpc_method(AddInput, Total)
    changed = rpc_event(Total)

The class is only a declaration; bind it to a client with client.rpc(Counter).

id class-attribute

id: str

methods class-attribute

methods: dict[str, RpcMethod[Any, Any]]

events class-attribute

events: dict[str, RpcEventDefinition[Any]]

event_type classmethod

event_type(event: RpcEventDefinition[Any] | str) -> str

The wire type of an event of this RPC: rpc.<id>.<name>.

rpc_method

rpc_method(
    input: None,
    output: type[_O],
    *,
    name: str | None = None,
    errors: Mapping[str, Any] | None = None,
) -> RpcMethod[None, _O]

rpc_method(
    input: type[_I],
    output: type[_O],
    *,
    name: str | None = None,
    errors: Mapping[str, Any] | None = None,
) -> RpcMethod[_I, _O]

rpc_method(
    input: Any,
    output: Any,
    *,
    name: str | None = None,
    errors: Mapping[str, Any] | None = None,
) -> RpcMethod[Any, Any]

Declares an RPC method on an :class:RpcDefinition.

Parameters:

Name	Type	Description	Default
input	Any	The input type, or None for a method without input.	required
output	Any	The output type. Use type(None) for a method that returns nothing.	required
name	str | None	The method name on the server. Defaults to the attribute name.	None
errors	Mapping[str, Any] | None	Data types of the errors the method declares, by error type. Matching error data is validated into these types on :attr:RpcCallError.data.	None

rpc_event

rpc_event(
    data: type[_T], *, name: str | None = None
) -> RpcEventDefinition[_T]

Declares an RPC event on an :class:RpcDefinition.

Parameters:

Name	Type	Description	Default
data	type[_T]	The type of the event's data.	required
name	str | None	The event name on the server. Defaults to the attribute name.	None

RpcMethod

RpcMethod(
    input: Any,
    output: Any,
    *,
    name: str | None,
    errors: Mapping[str, Any] | None,
)

Bases: Generic[_I, _O]

One method of an :class:RpcDefinition. Create it with :func:rpc_method.

name instance-attribute

name: str

input instance-attribute

input = input

output instance-attribute

output = output

errors instance-attribute

errors = dict(errors or {})

encode

encode(value: _I) -> object

Validates value against the input type and returns its JSON form.

decode

decode(rpc_id: str, value: object) -> _O

Validates a method's output.

error

error(
    rpc_id: str, exc: RpcError | RpcInternalError
) -> RpcCallError

Maps a transport error to :class:RpcCallError, decoding declared error data.

RpcEventDefinition

RpcEventDefinition(data: Any, *, name: str | None)

Bases: Generic[_T]

One event of an :class:RpcDefinition. Create it with :func:rpc_event.

name instance-attribute

name: str

data instance-attribute

data = data

parse

parse(event: RpcEvent) -> RpcEventPayload[_T]

Validates the data of a raw RPC event of this kind.

Raises:

Type	Description
ValidationError	The data does not match the declared type.

RpcClient

RpcClient(client: Opencode, definition: type[_D])

Bases: Generic[_D]

A plugin RPC bound to a synchronous client. Get one with client.rpc(Definition).

definition instance-attribute

definition = definition

call

call(
    method: RpcMethod[None, _O],
    input: None = None,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> _O

call(
    method: RpcMethod[_I, _O],
    input: _I,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> _O

call(
    method: RpcMethod[Any, Any],
    input: Any = None,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> Any

Calls a method of this RPC and returns its validated output.

Raises:

Type	Description
RpcCallError	The method failed.
RpcCallInternalError	The method crashed or returned invalid output.

events

events() -> Subscription[RpcEvent]

Subscribes to every event of this RPC, as raw events, on the client's shared event stream.

subscribe

subscribe(event: RpcEventDefinition[_T]) -> RpcEvents[_T]

Subscribes to one event of this RPC, with validated data, on the client's shared event stream.

AsyncRpcClient

AsyncRpcClient(
    client: AsyncOpencode, definition: type[_D]
)

Bases: Generic[_D]

A plugin RPC bound to an asynchronous client. Get one with client.rpc(Definition).

definition instance-attribute

definition = definition

call async

call(
    method: RpcMethod[None, _O],
    input: None = None,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> _O

call(
    method: RpcMethod[_I, _O],
    input: _I,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> _O

call(
    method: RpcMethod[Any, Any],
    input: Any = None,
    *,
    location: Location | NotGiven | None = NOT_GIVEN,
    extra_headers: Headers | None = None,
    timeout: Timeout | NotGiven = NOT_GIVEN,
) -> Any

Calls a method of this RPC and returns its validated output.

Raises:

Type	Description
RpcCallError	The method failed.
RpcCallInternalError	The method crashed or returned invalid output.

events

events() -> AsyncSubscription[RpcEvent]

Subscribes to every event of this RPC, as raw events, on the client's shared event stream.

subscribe

subscribe(
    event: RpcEventDefinition[_T],
) -> AsyncRpcEvents[_T]

Subscribes to one event of this RPC, with validated data, on the client's shared event stream.

RpcEvents

RpcEvents(
    source: Subscription[RpcEvent],
    event: RpcEventDefinition[_T],
)

Bases: Generic[_T]

Events of one kind from a plugin RPC. Iterate with for; close it, or use with, to leave.

get

get(timeout: float | None = None) -> RpcEventPayload[_T]

Returns the next event, waiting up to timeout seconds. See :meth:Subscription.get.

close

close() -> None

Leaves the shared event stream.

AsyncRpcEvents

AsyncRpcEvents(
    source: AsyncSubscription[RpcEvent],
    event: RpcEventDefinition[_T],
)

Bases: Generic[_T]

Events of one kind from a plugin RPC. Iterate with async for; close it, or use async with.

get async

get() -> RpcEventPayload[_T]

Returns the next event. See :meth:AsyncSubscription.get.

close async

close() -> None

Leaves the shared event stream.

RpcEventPayload dataclass

RpcEventPayload(name: str, data: _T, event: RpcEvent)

Bases: Generic[_T]

A plugin RPC event with validated data.

Attributes:

Name	Type	Description
name	str	The event name, without the rpc.<id>. prefix.
data	_T	The validated event data.
event	RpcEvent	The raw event, with its id, created time and location.

name instance-attribute

name: str

data instance-attribute

data: _T

event instance-attribute

event: RpcEvent

RpcCallError

RpcCallError(
    rpc_id: str,
    method: str,
    type: str,
    message: str,
    data: object = None,
)

Bases: OpencodeError

A plugin RPC method failed (the server answered with RpcError).

Attributes:

Name	Type	Description
rpc_id		The RPC that was called.
method		The method that was called.
type		The error type the plugin reported, for example "not_found".
message		The error message.
data		Error details, validated against the method's errors declaration when it names this type.

rpc_id instance-attribute

rpc_id = rpc_id

method instance-attribute

method = method

type instance-attribute

type = type

message instance-attribute

message = message

data instance-attribute

data = data

RpcCallInternalError

RpcCallInternalError(
    rpc_id: str,
    method: str,
    type: str,
    message: str,
    data: object = None,
)

Bases: RpcCallError

The plugin RPC crashed or returned output that does not match its declaration.

type is "rpc.internal" or "rpc.invalid_output". Output this client cannot validate against the method's declared output type is also reported as "rpc.invalid_output".