Skip to content

Channel API

@channel is a full-duplex session over Zenoh (and optionally WebSocket). The handler receives a ChannelSession. Callers use Istos.open_channel or @channel_client.

Details: Channels & Agent Sessions.

Bidirectional channels — full-duplex sessions for interactive agents.

A @handle replies once and a @stream yields many; a @channel keeps a session open so the handler can receive() inbound messages and send() outbound ones in any order — an agent that reads a turn, streams back tokens, then waits for the next. WebSocket is the transport at the HTTP edge; the handler still gets authorization, validation, DI and the request envelope, and can reach the rest of the mesh through the app.

@app.channel("agent/chat", ws="/chat")
async def chat(s: ChannelSession):
    await s.send({"role": "system", "text": "ready"})
    async for msg in s:
        async for tok in llm.stream(msg):
            await s.send(tok)
        await s.send({"done": True})

ChannelClosed

Bases: Exception

Raised by receive()/send() once the peer has hung up.

Source code in src/istos/primitives/channel.py 
34
35
class ChannelClosed(Exception):
    """Raised by receive()/send() once the peer has hung up."""

ChannelSession

The handler's end of a duplex session. The transport feeds inbound bytes and drains outbound ones; the handler works in terms of decoded messages.

Source code in src/istos/primitives/channel.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class ChannelSession:
    """The handler's end of a duplex session. The transport feeds inbound bytes
    and drains outbound ones; the handler works in terms of decoded messages."""

    def __init__(
        self,
        serializer: Serialize,
        send_sink: Callable[[bytes], Awaitable[None]],
        *,
        principal: Any = None,
        correlation_id: Optional[str] = None,
        attachment: Optional[bytes] = None,
        store: Any = None,
        conversation_id: Optional[str] = None,
    ) -> None:
        self._serializer = serializer
        self._send = send_sink
        self.principal = principal
        self.correlation_id = correlation_id
        self.attachment = attachment
        self._store = store
        self._conversation_id = conversation_id
        self._inbound: asyncio.Queue = asyncio.Queue()
        self._closed = False

    @property
    def conversation_id(self) -> Optional[str]:
        return self._conversation_id

    async def history(self, limit: int = 1000) -> list:
        """Prior messages for this conversation, oldest-first. Empty unless the
        channel is durable (``@channel(durable=True)``)."""
        if self._store is None or self._conversation_id is None:
            return []
        return list(await self._store.history(self._conversation_id, limit=limit))

    async def send(self, data: Any) -> None:
        """Push a message to the peer."""
        if self._closed:
            raise ChannelClosed("channel is closed")
        await self._send(self._serializer.serialize(data))
        if self._store is not None and self._conversation_id is not None:
            await self._store.append(self._conversation_id, "out", data)

    async def receive(self) -> Any:
        """Wait for the next message from the peer, or raise ChannelClosed."""
        item = await self._inbound.get()
        if item is _CLOSE:
            self._inbound.put_nowait(_CLOSE)  # keep every later receive() closed too
            raise ChannelClosed()
        data = self._serializer.deserialize(item)
        if self._store is not None and self._conversation_id is not None:
            await self._store.append(self._conversation_id, "in", data)
        return data

    def __aiter__(self) -> "ChannelSession":
        return self

    async def __anext__(self) -> Any:
        try:
            return await self.receive()
        except ChannelClosed:
            raise StopAsyncIteration

    # --- transport side ---

    def feed(self, raw: bytes) -> None:
        """Transport hook: hand an inbound message to the handler."""
        if not self._closed:
            self._inbound.put_nowait(raw)

    def close(self) -> None:
        """Transport hook: the peer is gone; unblock any pending receive()."""
        if not self._closed:
            self._closed = True
            self._inbound.put_nowait(_CLOSE)

    @property
    def closed(self) -> bool:
        return self._closed

close()

Transport hook: the peer is gone; unblock any pending receive().

Source code in src/istos/primitives/channel.py 
116
117
118
119
120
def close(self) -> None:
    """Transport hook: the peer is gone; unblock any pending receive()."""
    if not self._closed:
        self._closed = True
        self._inbound.put_nowait(_CLOSE)

feed(raw)

Transport hook: hand an inbound message to the handler.

Source code in src/istos/primitives/channel.py 
111
112
113
114
def feed(self, raw: bytes) -> None:
    """Transport hook: hand an inbound message to the handler."""
    if not self._closed:
        self._inbound.put_nowait(raw)

history(limit=1000) async

Prior messages for this conversation, oldest-first. Empty unless the channel is durable (@channel(durable=True)).

Source code in src/istos/primitives/channel.py 
74
75
76
77
78
79
async def history(self, limit: int = 1000) -> list:
    """Prior messages for this conversation, oldest-first. Empty unless the
    channel is durable (``@channel(durable=True)``)."""
    if self._store is None or self._conversation_id is None:
        return []
    return list(await self._store.history(self._conversation_id, limit=limit))

receive() async

Wait for the next message from the peer, or raise ChannelClosed.

Source code in src/istos/primitives/channel.py 
89
90
91
92
93
94
95
96
97
98
async def receive(self) -> Any:
    """Wait for the next message from the peer, or raise ChannelClosed."""
    item = await self._inbound.get()
    if item is _CLOSE:
        self._inbound.put_nowait(_CLOSE)  # keep every later receive() closed too
        raise ChannelClosed()
    data = self._serializer.deserialize(item)
    if self._store is not None and self._conversation_id is not None:
        await self._store.append(self._conversation_id, "in", data)
    return data

send(data) async

Push a message to the peer.

Source code in src/istos/primitives/channel.py 
81
82
83
84
85
86
87
async def send(self, data: Any) -> None:
    """Push a message to the peer."""
    if self._closed:
        raise ChannelClosed("channel is closed")
    await self._send(self._serializer.serialize(data))
    if self._store is not None and self._conversation_id is not None:
        await self._store.append(self._conversation_id, "out", data)

channel_wrapper

Registers a duplex handler and drives one session through it.

Source code in src/istos/primitives/channel.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class channel_wrapper:
    """Registers a duplex handler and drives one session through it."""

    def __init__(
        self,
        func: Callable,
        prefix: str,
        serializer: Serialize,
        *,
        authorizer: Optional[Authorizer] = None,
        exception_registry: Optional[ExceptionHandlerRegistry] = None,
        dependency_overrides: Optional[dict] = None,
        durable: bool = False,
        session_store: Any = None,
        middleware: Optional[MiddlewareStack] = None,
    ) -> None:
        if not inspect.iscoroutinefunction(func):
            raise TypeError(
                f"@channel requires an async function; {getattr(func, '__name__', func)!r} "
                "is not one."
            )
        self.func = func
        self.prefix = prefix
        self.serializer = serializer
        self.durable = durable
        self.session_store = session_store if durable else None
        self._authorizer = authorizer
        self._middleware = middleware
        self._exception_registry = exception_registry or get_default_registry()
        self._logger = get_logger("channel")

        params = list(inspect.signature(func).parameters.values())
        # The first positional parameter is the ChannelSession; the rest may be
        # network params or Depends(...).
        self._session_param = params[0].name if params else "session"
        self._depends_params = {
            p.name for p in params if extract_depends(p) is not None
        }
        self._has_depends = bool(self._depends_params)
        self._injected_params = set(self._depends_params) | {self._session_param}
        self._dependency_overrides = dependency_overrides or {}

    async def authorize(self, attachment: Optional[bytes], params: dict) -> Any:
        """Run the authorizer gate and return the principal (or raise). The fabric
        open handshake calls this to accept/deny before a session exists."""
        return await check_authorized(
            self._authorizer,
            AuthContext(
                prefix=self.prefix, key_expr=self.prefix, params=params,
                attachment=attachment, operation="channel",
            ),
        )

    async def run(
        self,
        session: ChannelSession,
        *,
        attachment: Optional[bytes],
        params: dict,
        principal: Any = _UNSET,
    ) -> None:
        """Validate, resolve DI, then run the handler for the life of the session.
        Authorizes first unless ``principal`` is supplied (already gated by the
        caller — e.g. the fabric open handshake)."""
        if principal is _UNSET:
            principal = await self.authorize(attachment, params)
        session.principal = principal

        req_ctx = get_request_context()
        req_ctx.prefix = self.prefix
        req_ctx.operation = "channel"
        req_ctx.principal = principal
        req_ctx.attachment = attachment
        env = RequestEnvelope.from_attachment(attachment)
        if env.correlation_id:
            req_ctx.correlation_id = env.correlation_id
        req_ctx.traceparent = env.traceparent
        session.correlation_id = req_ctx.correlation_id

        validated = validate_params(
            self.func, params, skip_params=self._injected_params
        )
        validated.pop("db", None)

        async with AsyncExitStack() as di_stack:
            call_kwargs = {self._session_param: session, **validated}
            if self._has_depends:
                call_kwargs = await resolve_dependencies(
                    self.func, call_kwargs, di_stack, cache={},
                    overrides=self._dependency_overrides,
                )

            async def _drive(_scope: Any = None) -> None:
                await self.func(**call_kwargs)

            # Middleware wraps the whole session — once at open, once at close.
            if self._middleware is not None:
                scope = RequestScope(
                    prefix=self.prefix, operation="channel", params=params,
                )
                scope.context.principal = req_ctx.principal
                scope.context.attachment = req_ctx.attachment
                scope.context.correlation_id = req_ctx.correlation_id
                scope.context.traceparent = req_ctx.traceparent
                await self._middleware.invoke(scope, _drive)
            else:
                await _drive()

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        """In-process invocation (TestClient) calls the handler directly."""
        return self.func(*args, **kwargs)

__call__(*args, **kwargs)

In-process invocation (TestClient) calls the handler directly.

Source code in src/istos/primitives/channel.py 
235
236
237
def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """In-process invocation (TestClient) calls the handler directly."""
    return self.func(*args, **kwargs)

authorize(attachment, params) async

Run the authorizer gate and return the principal (or raise). The fabric open handshake calls this to accept/deny before a session exists.

Source code in src/istos/primitives/channel.py 
169
170
171
172
173
174
175
176
177
178
async def authorize(self, attachment: Optional[bytes], params: dict) -> Any:
    """Run the authorizer gate and return the principal (or raise). The fabric
    open handshake calls this to accept/deny before a session exists."""
    return await check_authorized(
        self._authorizer,
        AuthContext(
            prefix=self.prefix, key_expr=self.prefix, params=params,
            attachment=attachment, operation="channel",
        ),
    )

run(session, *, attachment, params, principal=_UNSET) async

Validate, resolve DI, then run the handler for the life of the session. Authorizes first unless principal is supplied (already gated by the caller — e.g. the fabric open handshake).

Source code in src/istos/primitives/channel.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
async def run(
    self,
    session: ChannelSession,
    *,
    attachment: Optional[bytes],
    params: dict,
    principal: Any = _UNSET,
) -> None:
    """Validate, resolve DI, then run the handler for the life of the session.
    Authorizes first unless ``principal`` is supplied (already gated by the
    caller — e.g. the fabric open handshake)."""
    if principal is _UNSET:
        principal = await self.authorize(attachment, params)
    session.principal = principal

    req_ctx = get_request_context()
    req_ctx.prefix = self.prefix
    req_ctx.operation = "channel"
    req_ctx.principal = principal
    req_ctx.attachment = attachment
    env = RequestEnvelope.from_attachment(attachment)
    if env.correlation_id:
        req_ctx.correlation_id = env.correlation_id
    req_ctx.traceparent = env.traceparent
    session.correlation_id = req_ctx.correlation_id

    validated = validate_params(
        self.func, params, skip_params=self._injected_params
    )
    validated.pop("db", None)

    async with AsyncExitStack() as di_stack:
        call_kwargs = {self._session_param: session, **validated}
        if self._has_depends:
            call_kwargs = await resolve_dependencies(
                self.func, call_kwargs, di_stack, cache={},
                overrides=self._dependency_overrides,
            )

        async def _drive(_scope: Any = None) -> None:
            await self.func(**call_kwargs)

        # Middleware wraps the whole session — once at open, once at close.
        if self._middleware is not None:
            scope = RequestScope(
                prefix=self.prefix, operation="channel", params=params,
            )
            scope.context.principal = req_ctx.principal
            scope.context.attachment = req_ctx.attachment
            scope.context.correlation_id = req_ctx.correlation_id
            scope.context.traceparent = req_ctx.traceparent
            await self._middleware.invoke(scope, _drive)
        else:
            await _drive()

Cross-node channels over Zenoh — phase 2 of @channel.

A @channel handler can run on one node while a client (often a WebSocket gateway) opens a session from another. There is no Zenoh "connection", so a session is built from three keys under the channel prefix P and a session id S:

  • P/S/up — client → server messages (client puts, server subscribes)
  • P/S/down — server → client messages (server puts, client subscribes)
  • liveliness P/S — the client holds a token; when it drops (close or crash) the server tears the session down.

Opening is a one-shot query to P/S/open carrying the auth token as the attachment, so the authorizer runs before any session exists. On success both sides start pub/sub on up/down and the client declares its liveliness token.

ChannelClient

Caller's end of a fabric channel. Same surface as ChannelSession (send/receive/async-for/close), backed by Zenoh pub/sub.

Source code in src/istos/primitives/channel_fabric.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class ChannelClient:
    """Caller's end of a fabric channel. Same surface as ChannelSession
    (send/receive/async-for/close), backed by Zenoh pub/sub."""

    def __init__(
        self,
        session: Any,
        prefix: str,
        sid: str,
        serializer: Serialize,
        loop: asyncio.AbstractEventLoop,
        conversation_id: Optional[str] = None,
    ) -> None:
        self._session = session
        self._serializer = serializer
        self._loop = loop
        self.conversation_id = conversation_id
        self._up_key = f"{prefix}/{sid}/up"
        self._down_key = f"{prefix}/{sid}/down"
        self._live_key = f"{prefix}/{sid}"
        self._inbound: asyncio.Queue = asyncio.Queue()
        self._down_sub: Any = None
        self._live_token: Any = None
        self._closed = False

    def _subscribe_down(self) -> None:
        def cb(sample: zenoh.Sample) -> None:
            payload = bytes(sample.payload)
            self._loop.call_soon_threadsafe(self._inbound.put_nowait, payload)
        self._down_sub = self._session.declare_subscriber(self._down_key, cb)

    def _declare_liveliness(self) -> None:
        self._live_token = self._session.liveliness().declare_token(self._live_key)

    async def send(self, data: Any) -> None:
        if self._closed:
            raise ChannelClosed("channel is closed")
        raw = self._serializer.serialize(data)
        await asyncio.to_thread(self._session.put, self._up_key, raw)

    async def receive(self) -> Any:
        item = await self._inbound.get()
        if item is _CLOSE:
            self._inbound.put_nowait(_CLOSE)
            raise ChannelClosed()
        return self._serializer.deserialize(item)

    def __aiter__(self) -> "ChannelClient":
        return self

    async def __anext__(self) -> Any:
        try:
            return await self.receive()
        except ChannelClosed:
            raise StopAsyncIteration

    async def close(self) -> None:
        if self._closed:
            return
        self._closed = True
        # Dropping the liveliness token is what tells the server to tear down.
        if self._live_token is not None:
            with contextlib.suppress(Exception):
                self._live_token.undeclare()
        if self._down_sub is not None:
            with contextlib.suppress(Exception):
                self._down_sub.undeclare()
        self._inbound.put_nowait(_CLOSE)

    @property
    def closed(self) -> bool:
        return self._closed

FabricChannelServer

Serves one @channel over Zenoh: answers open handshakes, runs a handler per session, and tears sessions down when their liveliness token drops.

Source code in src/istos/primitives/channel_fabric.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
class FabricChannelServer:
    """Serves one @channel over Zenoh: answers open handshakes, runs a handler
    per session, and tears sessions down when their liveliness token drops."""

    def __init__(self, session: Any, wrapper: Any, loop: asyncio.AbstractEventLoop) -> None:
        self._session = session
        self._wrapper = wrapper
        self._loop = loop
        self._prefix = wrapper.prefix
        self._serializer: Serialize = wrapper.serializer
        self._sessions: Dict[str, _ServerSession] = {}
        self._queryable: Any = None
        self._live_sub: Any = None
        self._logger = get_logger("channel")

    def bind(self) -> None:
        self._queryable = self._session.declare_queryable(
            f"{self._prefix}/*/open", self._on_open, complete=True
        )
        self._live_sub = self._session.liveliness().declare_subscriber(
            f"{self._prefix}/*", self._on_liveliness, history=False
        )

    def unbind(self) -> None:
        for sid in list(self._sessions):
            self._teardown(sid)
        if self._queryable is not None:
            with contextlib.suppress(Exception):
                self._queryable.undeclare()
        if self._live_sub is not None:
            with contextlib.suppress(Exception):
                self._live_sub.undeclare()

    # --- open handshake (Zenoh queryable thread) ---

    def _on_open(self, query: zenoh.Query) -> None:
        key = str(query.selector.key_expr)
        sid = _sid_of(key, self._prefix)
        attachment = _attachment_of(query)
        params: dict = {}
        if getattr(query.selector, "parameters", None):
            params = decode_params(
                dict(cast(Iterable[Tuple[str, str]], query.selector.parameters))
            )

        fut = asyncio.run_coroutine_threadsafe(
            self._open(sid, attachment, params), self._loop
        )
        try:
            reply = fut.result(timeout=10)
        except Exception as e:  # pragma: no cover - defensive
            reply = reply_err(str(e))
        with contextlib.suppress(Exception):
            query.reply(key, self._serializer.serialize(reply))

    async def _open(self, sid: str, attachment: Optional[bytes], params: dict) -> dict:
        conversation_id = params.pop("conversation_id", None)
        try:
            principal = await self._wrapper.authorize(attachment, params)
        except UnauthorizedError as e:
            return reply_err(str(e), code="unauthorized")

        down_key = f"{self._prefix}/{sid}/down"

        async def sink(raw: bytes) -> None:
            await asyncio.to_thread(self._session.put, down_key, raw)

        chan = ChannelSession(
            self._serializer, sink, principal=principal, attachment=attachment,
            store=self._wrapper.session_store, conversation_id=conversation_id,
        )

        def up_cb(sample: zenoh.Sample) -> None:
            payload = bytes(sample.payload)
            self._loop.call_soon_threadsafe(chan.feed, payload)

        up_sub = self._session.declare_subscriber(f"{self._prefix}/{sid}/up", up_cb)
        self._sessions[sid] = _ServerSession(chan, up_sub)
        self._loop.create_task(self._run(sid, chan, attachment, params, principal))
        return {"ok": True, "sid": sid}

    async def _run(
        self, sid: str, chan: ChannelSession, attachment: Optional[bytes],
        params: dict, principal: Any,
    ) -> None:
        try:
            await self._wrapper.run(
                chan, attachment=attachment, params=params, principal=principal
            )
        except Exception as e:
            self._logger.error(
                "Channel session %s on %s failed: %s", sid, self._prefix, e,
                exc_info=True, extra={"prefix": self._prefix},
            )
        finally:
            self._teardown(sid)

    # --- liveliness teardown (Zenoh subscriber thread) ---

    def _on_liveliness(self, sample: zenoh.Sample) -> None:
        if sample.kind == zenoh.SampleKind.DELETE:
            sid = _sid_of(str(sample.key_expr), self._prefix)
            self._loop.call_soon_threadsafe(self._teardown, sid)

    def _teardown(self, sid: str) -> None:
        state = self._sessions.pop(sid, None)
        if state is None:
            return
        state.session.close()  # unblocks the handler's receive() → it returns
        with contextlib.suppress(Exception):
            state.up_sub.undeclare()

is_error_payload(parsed)

Whether a decoded reply is an :class:ErrorResponse wire payload.

A handler that raises replies with an envelope rather than sending an exception, and the envelope answers .get() like any other dict, so an unchecked caller reads a failure as data.

query_once, @query, stream_query and open_channel check this themselves. Use it directly for replies you decode yourself, and for multi-reply results, which are passed through unchecked. Pair it with :func:error_from_payload to raise.

Detection prefers the :data:ERROR_MARKER discriminator: present and truthy is an error, present and falsy is a normal result — the escape hatch for a handler whose success value legitimately carries error/code/ message. When the marker is absent (an old responder, or a client in another language), fall back to the legacy rule: all three of error, code and message present.

Source code in src/istos/errors.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def is_error_payload(parsed: Any) -> bool:
    """Whether a decoded reply is an :class:`ErrorResponse` wire payload.

    A handler that raises replies with an envelope rather than sending an
    exception, and the envelope answers ``.get()`` like any other dict, so an
    unchecked caller reads a failure as data.

    ``query_once``, ``@query``, ``stream_query`` and ``open_channel`` check this
    themselves. Use it directly for replies you decode yourself, and for
    multi-reply results, which are passed through unchecked. Pair it with
    :func:`error_from_payload` to raise.

    Detection prefers the :data:`ERROR_MARKER` discriminator: present and truthy
    is an error, present and falsy is a normal result — the escape hatch for a
    handler whose success value legitimately carries ``error``/``code``/
    ``message``. When the marker is absent (an old responder, or a client in
    another language), fall back to the legacy rule: all three of ``error``,
    ``code`` and ``message`` present.
    """
    if not isinstance(parsed, dict):
        return False
    marker = parsed.get(ERROR_MARKER)
    if marker is not None:
        return bool(marker)
    return all(field in parsed for field in ("error", "code", "message"))

Client-side decorators for reaching @stream and @channel — the declarative counterparts to @query (which reaches @handle) and @subscribe (which reaches @publish). A service is a mix of senders and receivers; these let the receiving side be attached the same way, on the app or a router.

@stream_client hands the body the live chunk iterator; @channel_client hands it an open ChannelClient and closes it when the body returns. Call kwargs become the selector params, and a per-call token= (or a decorator-level token=) carries the auth token — just like @query.

channel_client_wrapper

Bases: _client_base

Reaches a @channel: opens a session, passes the ChannelClient to the body, and closes it when the body returns.

Source code in src/istos/primitives/clients.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class channel_client_wrapper(_client_base):
    """Reaches a @channel: opens a session, passes the ChannelClient to the body,
    and closes it when the body returns."""

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        token = kwargs.pop("token", None)
        chan = await self._app.open_channel(
            self.prefix, token=token if token is not None else self._token,
            timeout_s=self.timeout_s, serializer=self.serializer, **kwargs,
        )
        try:
            return await self._drive(args, chan)
        finally:
            await chan.close()

stream_client_wrapper

Bases: _client_base

Reaches a @stream: opens the stream and passes its chunk iterator to the body. async for chunk in it inside the body.

Source code in src/istos/primitives/clients.py 
73
74
75
76
77
78
79
80
81
82
83
class stream_client_wrapper(_client_base):
    """Reaches a @stream: opens the stream and passes its chunk iterator to the
    body. ``async for chunk in it`` inside the body."""

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        token = kwargs.pop("token", None)
        stream = self._app.stream_query(
            self.prefix, timeout_s=self.timeout_s, serializer=self.serializer,
            token=token if token is not None else self._token, **kwargs,
        )
        return await self._drive(args, stream)