Shared event stream and typed plugin RPC

CHANGELOG.md

@@ -3,6 +3,15 @@
 Releases are published automatically on every merge to `main`. Release notes for each version are on the
 [GitHub releases page](https://github.com/shrijayan/opencode-python-sdk/releases).
 
+## Unreleased
+
+- `client.events`: a shared event stream with one connection for all subscribers, bounded per-subscriber queues,
+  `server.connected` replay, `of_type` / `for_session` / `where` filters, `on_activity` keepalive hook, cancellation,
+  and opt-in reconnection with backoff (`EventHub(client, reconnect=True)`).
+- `client.rpc(Definition)`: typed plugin RPC calls and event subscriptions declared with `RpcDefinition`,
+  `rpc_method` and `rpc_event`; RPC failures raise `RpcCallError` / `RpcCallInternalError`.
+- `Stream.on_activity` and `Stream.interrupt()`.
+
 ## 0.1.0
 
 First release, generated from OpenCode 2.1.2: sync and async clients for all 142 operations, typed models and

README.md

@@ -169,6 +169,82 @@ Events are typed (`SessionCreatedEvent`, `SessionTextDeltaEvent`, and so on). Ev
 not know arrive as `UnknownEvent` with all their fields kept. Streams apply only the connect timeout by default, since
 they stay open between events.
 
+### Shared event stream
+
+`client.events` shares one connection among every subscriber, like `client.event` in the TypeScript client. The
+connection opens with the first subscriber and closes when the last one leaves. Each subscriber has its own queue of
+up to 4096 events; one that falls further behind fails with `EventOverflowError` without affecting the others. The
+latest `server.connected` event is replayed to subscribers that join later.
+
+```python
+from opencode.types import SessionIdleEvent
+
+with client.events.of_type(SessionIdleEvent).for_session(session.id) as idle:
+    client.session.prompt(session.id, text="Hello")
+    event = idle.get(timeout=300)  # typed as SessionIdleEvent
+```
+
+- Filters chain: `of_type(*classes)` narrows the yielded type, `for_session(id)` keeps one session's events, and
+  `where(predicate)` takes any test. Entering the `with` block joins the stream, so nothing published after that is
+  missed.
+- `subscribe(on_activity=callback)` calls `callback` for every chunk received, keepalive comments included, as a
+  liveness signal.
+- `close()` (or leaving the `with` block) ends the subscription, also from another thread. In async code, cancel a wait
+  with `anyio.fail_after`, `asyncio.timeout` or a trio cancel scope; the subscription stays usable.
+- Reconnection is opt-in: `EventHub(client, reconnect=True)` (or `AsyncEventHub`, or a `Reconnect(...)` policy) reopens
+  the connection with exponential backoff after connection errors, 5xx and 429. Subscribers see `server.connected`
+  once. **The stream is volatile**: events published while disconnected are lost, so re-read state through the API
+  after a reconnect when it matters.
+
+The async hub works the same way under asyncio and trio:
+
+```python
+async with client.events.for_session(session.id) as events:
+    async for event in events:
+        print(event.type)
+```
+
+`client.event.subscribe()` still opens a dedicated connection.
+
+### Plugin RPC
+
+Declare a plugin RPC once as a class, then call it with typed inputs and outputs:
+
+```python
+from pydantic import BaseModel
+from opencode import RpcCallError, RpcDefinition, rpc_event, rpc_method
+
+
+class AddInput(BaseModel):
+    amount: int
+
+
+class Total(BaseModel):
+    total: int
+
+
+class Counter(RpcDefinition, id="counter"):
+    add = rpc_method(AddInput, Total)
+    reset = rpc_method(None, Total)
+    changed = rpc_event(Total)
+
+
+counter = client.rpc(Counter)
+try:
+    print(counter.call(Counter.add, AddInput(amount=2)).total)
+except RpcCallError as error:
+    print(error.type, error.message, error.data)
+
+with counter.subscribe(Counter.changed) as changes:  # only rpc.counter.changed, with validated data
+    for change in changes:
+        print(change.data.total)
+```
+
+Any type pydantic validates works for inputs, outputs and event data. `rpc_method(..., errors={"not_found": Model})`
+validates the `data` of declared errors. Failures raise `RpcCallError`; crashes and outputs that do not match the
+declared type raise `RpcCallInternalError`. `counter.events()` yields every raw event of the RPC. The untyped
+`client.rpc.call(rpc_id, method, input=...)` is still available.
+
 ## Terminals
 
 `client.pty.attach()` opens an interactive connection to a PTY over a WebSocket. It needs the `pty` extra:

docs/guides/events.md

@@ -47,3 +47,42 @@ async with await client.event.subscribe() as events:
     async for event in events:
         print(event.type)
 ```
+
+## Shared event stream
+
+`client.events` shares one connection among every subscriber, like `client.event` in the TypeScript client. The
+connection opens with the first subscriber and closes when the last one leaves. Each subscriber has its own queue of
+up to 4096 events; one that falls further behind fails with `EventOverflowError` without affecting the others. The
+latest `server.connected` event is replayed to subscribers that join later.
+
+```python
+from opencode.types import SessionIdleEvent
+
+with client.events.of_type(SessionIdleEvent).for_session(session.id) as idle:
+    client.session.prompt(session.id, text="Hello")
+    event = idle.get(timeout=300)  # typed as SessionIdleEvent
+```
+
+- Filters chain: `of_type(*classes)` narrows the yielded type, `for_session(id)` keeps one session's events, and
+  `where(predicate)` takes any test. Entering the `with` block joins the stream, so nothing published after that is
+  missed.
+- `subscribe(on_activity=callback)` calls `callback` for every chunk received, keepalive comments included, as a
+  liveness signal.
+- `close()` (or leaving the `with` block) ends the subscription, also from another thread. In async code, cancel a wait
+  with `anyio.fail_after`, `asyncio.timeout` or a trio cancel scope; the subscription stays usable.
+- Reconnection is opt-in: `EventHub(client, reconnect=True)` (or `AsyncEventHub`, or a `Reconnect(...)` policy) reopens
+  the connection with exponential backoff after connection errors, 5xx and 429. Subscribers see `server.connected`
+  once. **The stream is volatile** (see [Delivery](#delivery)): events published while disconnected are lost, so re-read state through the API
+  after a reconnect when it matters.
+
+The async hub works the same way under asyncio and trio:
+
+```python
+async with client.events.for_session(session.id) as events:
+    async for event in events:
+        print(event.type)
+```
+
+`client.event.subscribe()` still opens a dedicated connection.
+
+Typed plugin RPC events are covered in the [plugin RPC guide](plugin-rpc.md).

docs/guides/index.md

@@ -28,7 +28,7 @@ entry; the common ones also have a guide. CI fails if a group on the client is m
 | `client.provider` | Model providers | - | [`ProviderResource`](../reference/resources.md#opencode.resources.provider.ProviderResource) |
 | `client.pty` | Terminals | [Interactive terminals](terminals.md) | [`PtyResource`](../reference/resources.md#opencode.resources.pty.PtyResource) |
 | `client.reference` | References | - | [`ReferenceResource`](../reference/resources.md#opencode.resources.reference.ReferenceResource) |
-| `client.rpc` | Plugin RPC | - | [`RpcResource`](../reference/resources.md#opencode.resources.rpc.RpcResource) |
+| `client.rpc` | Plugin RPC | [Plugin RPC](plugin-rpc.md) | [`RpcResource`](../reference/resources.md#opencode.resources.rpc.RpcResource) |
 | `client.server` | Server info and health | - | [`ServerResource`](../reference/resources.md#opencode.resources.server.ServerResource) |
 | `client.session` | Sessions, prompts and their sub-resources | [Guide](sessions.md) | [`SessionResource`](../reference/resources.md#opencode.resources.session.SessionResource) |
 | `client.shell` | Shells | - | [`ShellResource`](../reference/resources.md#opencode.resources.shell.ShellResource) |

docs/guides/plugin-rpc.md

@@ -0,0 +1,53 @@
+# Plugin RPC
+
+Declare a plugin RPC once as a class, then call it with typed inputs and outputs:
+
+```python
+from pydantic import BaseModel
+from opencode import RpcCallError, RpcDefinition, rpc_event, rpc_method
+
+
+class AddInput(BaseModel):
+    amount: int
+
+
+class Total(BaseModel):
+    total: int
+
+
+class Counter(RpcDefinition, id="counter"):
+    add = rpc_method(AddInput, Total)
+    reset = rpc_method(None, Total)
+    changed = rpc_event(Total)
+
+
+counter = client.rpc(Counter)
+try:
+    print(counter.call(Counter.add, AddInput(amount=2)).total)
+except RpcCallError as error:
+    print(error.type, error.message, error.data)
+
+with counter.subscribe(Counter.changed) as changes:  # only rpc.counter.changed, with validated data
+    for change in changes:
+        print(change.data.total)
+```
+
+Any type pydantic validates works for inputs, outputs and event data. `rpc_method(..., errors={"not_found": Model})`
+validates the `data` of declared errors. Failures raise `RpcCallError`; crashes and outputs that do not match the
+declared type raise `RpcCallInternalError`. `counter.events()` yields every raw event of the RPC. The untyped
+`client.rpc.call(rpc_id, method, input=...)` is still available.
+
+
+## Async
+
+The async client returns `AsyncRpcClient`, with the same methods:
+
+```python
+counter = client.rpc(Counter)
+total = await counter.call(Counter.add, AddInput(amount=2))
+async with counter.subscribe(Counter.changed) as changes:
+    async for change in changes:
+        print(change.data.total)
+```
+
+RPC events travel on the [shared event stream](events.md#shared-event-stream), so they share its delivery rules.

docs/reference/events.md

@@ -0,0 +1,50 @@
+# 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](../guides/events.md) and the [plugin RPC guide](../guides/plugin-rpc.md).
+
+## Event hubs
+
+::: opencode.EventHub
+
+::: opencode.AsyncEventHub
+
+::: opencode.Subscription
+
+::: opencode.AsyncSubscription
+
+::: opencode.Reconnect
+
+::: opencode.EventOverflowError
+
+::: opencode.SUBSCRIBER_CAPACITY
+
+::: opencode.session_id_of
+
+## Plugin RPC
+
+::: opencode._rpc.TypedRpcResource
+
+::: opencode.RpcDefinition
+
+::: opencode.rpc_method
+
+::: opencode.rpc_event
+
+::: opencode.RpcMethod
+
+::: opencode.RpcEventDefinition
+
+::: opencode.RpcClient
+
+::: opencode.AsyncRpcClient
+
+::: opencode.RpcEvents
+
+::: opencode.AsyncRpcEvents
+
+::: opencode.RpcEventPayload
+
+::: opencode.RpcCallError
+
+::: opencode.RpcCallInternalError

docs/reference/index.md

@@ -8,6 +8,7 @@ Generated from the docstrings in `src/opencode`.
 | [Server launcher](server.md) | `OpencodeServer`, `AsyncOpencodeServer` |
 | [Resources](resources.md) | Every `client.<group>` and its methods |
 | [Pagination and streams](pagination.md) | `SyncCursorPage`, `AsyncPaginator`, `Stream`, `AsyncStream` |
+| [Shared events and RPC](events.md) | `EventHub`, `Subscription`, `RpcDefinition`, `RpcClient` |
 | [Exceptions](exceptions.md) | Status errors and every declared server error |
 | [Types](types/index.md) | The models in `opencode.types`, grouped by domain |
 

mkdocs.yml

@@ -139,6 +139,7 @@ nav:
       - Models and unions: guides/models.md
       - Pagination: guides/pagination.md
       - Events and streaming: guides/events.md
+      - Plugin RPC: guides/plugin-rpc.md
       - Interactive terminals: guides/terminals.md
       - Errors: guides/errors.md
       - Retries and timeouts: guides/retries.md
@@ -151,6 +152,7 @@ nav:
       - Server launcher: reference/server.md
       - Resources: reference/resources.md
       - Pagination and streams: reference/pagination.md
+      - Shared events and RPC: reference/events.md
       - Terminals: reference/terminals.md
       - Exceptions: reference/exceptions.md
       - Types: reference/types/index.md

src/opencode/__init__.py

@@ -81,6 +81,33 @@
 from ._types import NOT_GIVEN, NotGiven
 from ._version import __version__
 
+# isort: split
+# Hand-written helpers, kept apart from the generated error-import block above.
+from ._events import (
+    SUBSCRIBER_CAPACITY,
+    AsyncEventHub,
+    AsyncSubscription,
+    EventHub,
+    EventOverflowError,
+    Reconnect,
+    Subscription,
+    session_id_of,
+)
+from ._rpc import (
+    AsyncRpcClient,
+    AsyncRpcEvents,
+    RpcCallError,
+    RpcCallInternalError,
+    RpcClient,
+    RpcDefinition,
+    RpcEventDefinition,
+    RpcEventPayload,
+    RpcEvents,
+    RpcMethod,
+    rpc_event,
+    rpc_method,
+)
+
 # <generated:api-version>
 API_VERSION = "2.1.2"
 """The OpenCode server version this client was generated from."""
@@ -123,6 +150,27 @@
     "UnprocessableEntityError",
     "__version__",
     "types",
+    # Shared event stream and typed plugin RPC
+    "SUBSCRIBER_CAPACITY",
+    "AsyncEventHub",
+    "AsyncRpcClient",
+    "AsyncRpcEvents",
+    "AsyncSubscription",
+    "EventHub",
+    "EventOverflowError",
+    "Reconnect",
+    "RpcCallError",
+    "RpcCallInternalError",
+    "RpcClient",
+    "RpcDefinition",
+    "RpcEventDefinition",
+    "RpcEventPayload",
+    "RpcEvents",
+    "RpcMethod",
+    "Subscription",
+    "rpc_event",
+    "rpc_method",
+    "session_id_of",
     # <generated:error-exports>
     "AgentNotFoundError",
     "CommandExecutionError",

src/opencode/_client.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+from functools import cached_property
 from typing import TYPE_CHECKING
 
 from typing_extensions import Self
 
 from ._base_client import AsyncAPIClient, SyncAPIClient
 from ._constants import DEFAULT_MAX_RETRIES, DEFAULT_TIMEOUT, DEFAULT_USERNAME
+from ._events import AsyncEventHub, EventHub
+from ._rpc import AsyncTypedRpcResource, TypedRpcResource
 from ._types import NOT_GIVEN, NotGiven, Timeout
 from .resources import AsyncResources, SyncResources
 
@@ -70,6 +73,16 @@ def __init__(
             http_client=http_client,
         )
 
+    @cached_property
+    def events(self) -> EventHub:
+        """The shared event stream: one connection for every subscriber. See :class:`opencode.EventHub`."""
+        return EventHub(self)
+
+    @cached_property
+    def rpc(self) -> TypedRpcResource:
+        """Plugin RPC calls: ``client.rpc.call(...)``, or ``client.rpc(Definition)`` for typed access."""
+        return TypedRpcResource(self)
+
     def with_options(
         self,
         *,
@@ -138,6 +151,16 @@ def __init__(
             http_client=http_client,
         )
 
+    @cached_property
+    def events(self) -> AsyncEventHub:
+        """The shared event stream: one connection for every subscriber. See :class:`opencode.AsyncEventHub`."""
+        return AsyncEventHub(self)
+
+    @cached_property
+    def rpc(self) -> AsyncTypedRpcResource:
+        """Plugin RPC calls: ``client.rpc.call(...)``, or ``client.rpc(Definition)`` for typed access."""
+        return AsyncTypedRpcResource(self)
+
     def with_options(
         self,
         *,