teff.channels

teff.channels

Channel adapters: run one workflow.yaml over many transports.

Constitution Principle IX: observability and a single source of truth. The channel layer keeps the workflow YAML as the one executable spec and binds transport adapters (HTTP/SSE, Telegram, generic webhooks) onto the same durable :class:~teff.assistant.Assistant service, so interrupt handling, checkpoints and message history behave identically on every surface.

Importing this package is dependency-free (stdlib + httpx): the HTTP adapter needs the optional teff[channels] extra and is imported lazily via :func:create_http_app.

Public API::

assistant = build_assistant("workflow.yaml")          # one durable service
hook      = build_webhook(assistant, spec)            # generic webhook
bot       = TelegramChannel(assistant, token=...)     # polling/webhook
app       = create_http_app(assistant)                # FastAPI + SSE
router    = create_http_router(assistant)             # mount into an app

Modules:

Name	Description
factory	Build the durable Assistant service from a workflow YAML file.
http	HTTP/SSE channel: serve one durable Assistant over FastAPI.
reply	Extract the final assistant reply from a completed turn and shape the
telegram	Telegram channel: run a workflow from Telegram messages.
webhook	Generic webhook channel: run a workflow on any inbound JSON payload.

Classes:

Name	Description
TelegramChannel	A Telegram Bot API adapter over a shared Assistant.
WebhookChannel	One inbound webhook route bound to a shared Assistant.

Functions:

Name	Description
HTTPChannel	Build an :class:~teff.channels.http.HTTPChannel (needs teff[channels]).
build_assistant	Compile path into a durable, interrupt-aware :class:Assistant.
build_webhook	Build one generic webhook channel from a channels.webhook entry.
create_http_app	Build the HTTP/SSE FastAPI app for assistant (needs teff[channels]).
create_http_router	Build the HTTP/SSE routes for assistant as a mountable APIRouter.
load_channels	Return the parsed channels: block of path ({} when absent).
reply_from_state	Extract the best-effort assistant reply from a finished state.
reply_text	Return the best-effort assistant reply for result.
turn_response	Shape one turn into the channel response format.

TelegramChannel

A Telegram Bot API adapter over a shared Assistant.

Methods:

Name	Description
handle_update	Process one Telegram update: run a turn and reply in-chat.
run	Long-poll for updates forever (or a single pass with once).
send_message	Send a plain text reply (interrupt prompts included).
session_id_for	Telegram chats map one-to-one to durable sessions.
set_webhook	Point Telegram at url (call once, then serve the POSTs).

Source code in teff/channels/telegram.py

class TelegramChannel:
    """A Telegram Bot API adapter over a shared ``Assistant``."""

    def __init__(
        self,
        assistant: Assistant,
        token: str,
        *,
        owner: str = "telegram",
        poll_timeout: int = 30,
    ):
        self.assistant = assistant
        self.token = token
        self.owner = owner
        self.poll_timeout = poll_timeout
        self._base = API_BASE + token
        self._offset: int | None = None

    def session_id_for(self, chat_id: int | str) -> str:
        """Telegram chats map one-to-one to durable sessions."""
        return f"tg-{chat_id}"

    async def _api(self, method: str, **params: Any) -> dict:
        async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
            resp = await client.post(f"{self._base}/{method}", json=params)
            resp.raise_for_status()
            data = resp.json()
            if not data.get("ok"):
                raise RuntimeError(f"telegram {method} failed: {data}")
            return data["result"]

    async def send_message(self, chat_id: int, text: str) -> None:
        """Send a plain text reply (interrupt prompts included)."""
        await self._api("sendMessage", chat_id=chat_id, text=text)

    async def handle_update(self, update: dict[str, Any]) -> None:
        """Process one Telegram update: run a turn and reply in-chat.

        The checkpoint owner is the sender's Telegram user id
        (``message.from.id``), so every user's sessions are isolated.
        """
        message = update.get("message") or update.get("edited_message")
        if not message:
            return
        chat_id = message["chat"]["id"]
        text = message.get("text")
        if not text:
            return
        owner = str(message.get("from", {}).get("id") or chat_id)
        session_id = self.session_id_for(chat_id)
        result = await self.assistant.run(session_id, text, owner=owner)
        if result.waiting:
            prompt = result.prompt or "?"
            await self.send_message(chat_id, f"⏳ {prompt}")
        else:
            from teff.channels.reply import reply_text

            await self.send_message(chat_id, reply_text(result))

    async def run(self, *, once: bool = False) -> None:
        """Long-poll for updates forever (or a single pass with ``once``)."""
        logger.info("telegram channel polling for updates")
        while True:
            params: dict[str, Any] = {"timeout": self.poll_timeout}
            if self._offset is not None:
                params["offset"] = self._offset
            updates = await self._api("getUpdates", **params)
            for update in updates:
                self._offset = int(update["update_id"]) + 1
                try:
                    await self.handle_update(update)
                except Exception:
                    logger.exception("error handling telegram update")
            if once:
                return
            await asyncio.sleep(0.1)

    async def set_webhook(self, url: str) -> None:
        """Point Telegram at ``url`` (call once, then serve the POSTs)."""
        await self._api("setWebhook", url=url)

    async def delete_webhook(self) -> None:
        await self._api("deleteWebhook")

handle_update async

handle_update(update)

Process one Telegram update: run a turn and reply in-chat.

The checkpoint owner is the sender's Telegram user id (message.from.id), so every user's sessions are isolated.

Source code in teff/channels/telegram.py

async def handle_update(self, update: dict[str, Any]) -> None:
    """Process one Telegram update: run a turn and reply in-chat.

    The checkpoint owner is the sender's Telegram user id
    (``message.from.id``), so every user's sessions are isolated.
    """
    message = update.get("message") or update.get("edited_message")
    if not message:
        return
    chat_id = message["chat"]["id"]
    text = message.get("text")
    if not text:
        return
    owner = str(message.get("from", {}).get("id") or chat_id)
    session_id = self.session_id_for(chat_id)
    result = await self.assistant.run(session_id, text, owner=owner)
    if result.waiting:
        prompt = result.prompt or "?"
        await self.send_message(chat_id, f"⏳ {prompt}")
    else:
        from teff.channels.reply import reply_text

        await self.send_message(chat_id, reply_text(result))

run async

run(*, once=False)

Long-poll for updates forever (or a single pass with once).

Source code in teff/channels/telegram.py

async def run(self, *, once: bool = False) -> None:
    """Long-poll for updates forever (or a single pass with ``once``)."""
    logger.info("telegram channel polling for updates")
    while True:
        params: dict[str, Any] = {"timeout": self.poll_timeout}
        if self._offset is not None:
            params["offset"] = self._offset
        updates = await self._api("getUpdates", **params)
        for update in updates:
            self._offset = int(update["update_id"]) + 1
            try:
                await self.handle_update(update)
            except Exception:
                logger.exception("error handling telegram update")
        if once:
            return
        await asyncio.sleep(0.1)

send_message async

send_message(chat_id, text)

Send a plain text reply (interrupt prompts included).

Source code in teff/channels/telegram.py

async def send_message(self, chat_id: int, text: str) -> None:
    """Send a plain text reply (interrupt prompts included)."""
    await self._api("sendMessage", chat_id=chat_id, text=text)

session_id_for

session_id_for(chat_id)

Telegram chats map one-to-one to durable sessions.

Source code in teff/channels/telegram.py

def session_id_for(self, chat_id: int | str) -> str:
    """Telegram chats map one-to-one to durable sessions."""
    return f"tg-{chat_id}"

set_webhook async

set_webhook(url)

Point Telegram at url (call once, then serve the POSTs).

Source code in teff/channels/telegram.py

async def set_webhook(self, url: str) -> None:
    """Point Telegram at ``url`` (call once, then serve the POSTs)."""
    await self._api("setWebhook", url=url)

WebhookChannel

One inbound webhook route bound to a shared Assistant.

Methods:

Name	Description
handle	Validate payload, run one turn, return the channel response.
message_for	Render the one-turn message from the payload fields.
owner_for	Resolve the checkpoint owner from the configured owner spec.
session_id_for	Derive the durable session id from the payload.
validate	Return schema errors for payload (empty when valid).

Source code in teff/channels/webhook.py

class WebhookChannel:
    """One inbound webhook route bound to a shared ``Assistant``."""

    def __init__(self, assistant: Assistant, spec: dict[str, Any]):
        self.assistant = assistant
        self.spec = spec
        self.path = spec.get("path", "/webhook")
        self.schema = spec.get("schema") or {}
        self.session_key = spec.get("session_key")
        self.message_template = spec.get("input", {}).get("message", "{message}")
        self.max_iterations = spec.get("max_iterations", 80)
        self.owner_spec = spec.get("owner") or "default"

    def session_id_for(self, payload: dict[str, Any]) -> str:
        """Derive the durable session id from the payload.

        Uses ``session_key`` when configured; otherwise a content hash, so
        the same payload always resumes the same conversation.
        """
        if self.session_key is not None:
            value = payload.get(self.session_key)
            if value is not None:
                return str(value)
        raw = str(payload).encode("utf-8", "replace")
        return "wh-" + hashlib.sha1(raw).hexdigest()[:24]

    def message_for(self, payload: dict[str, Any]) -> str:
        """Render the one-turn ``message`` from the payload fields."""
        return render_template(self.message_template, payload)

    def owner_for(
        self,
        payload: Mapping[str, Any],
        headers: Mapping[str, Any] | None = None,
    ) -> str:
        """Resolve the checkpoint owner from the configured ``owner`` spec.

        ``payload.<field>`` reads the body, ``header.<Name>`` reads a
        request header (case-insensitive), ``fixed:<value>`` is a constant,
        and anything else falls back to the spec verbatim (``default``).
        """
        spec = self.owner_spec
        if isinstance(spec, str) and spec.startswith("payload."):
            return str(payload.get(spec[len("payload.") :], "default"))
        if isinstance(spec, str) and spec.startswith("header."):
            name = spec[len("header.") :].lower()
            if headers:
                for key, value in headers.items():
                    if str(key).lower() == name and value is not None:
                        return str(value)
            return "default"
        if isinstance(spec, str) and spec.startswith("fixed:"):
            return spec[len("fixed:") :]
        return str(spec)

    def validate(self, payload: Any) -> list[str]:
        """Return schema errors for *payload* (empty when valid)."""
        if not self.schema:
            return []
        if not isinstance(payload, dict):
            return ["payload must be an object"]
        return validate_json(payload, self.schema)

    async def handle(
        self,
        payload: dict[str, Any],
        *,
        owner: str | None = None,
        headers: Mapping[str, Any] | None = None,
    ) -> dict:
        """Validate *payload*, run one turn, return the channel response.

        *owner* overrides the configured ``owner:`` spec (the CLI passes the
        resolved value when it wants to override).  The return value matches
        the HTTP channel's shape: ``ok`` plus a turn of ``session_id`` /
        ``waiting`` / ``message`` (the reply, or the interrupt prompt when
        ``waiting``).
        """
        errors = self.validate(payload)
        if errors:
            return {"ok": False, "errors": errors}
        session_id = self.session_id_for(payload)
        result = await self.assistant.run(
            session_id,
            self.message_for(payload),
            owner=owner or self.owner_for(payload, headers),
            max_iterations=self.max_iterations,
        )
        return {"ok": True, **turn_response(result, session_id)}

handle async

handle(payload, *, owner=None, headers=None)

Validate payload, run one turn, return the channel response.

owner overrides the configured owner: spec (the CLI passes the resolved value when it wants to override). The return value matches the HTTP channel's shape: ok plus a turn of session_id / waiting / message (the reply, or the interrupt prompt when waiting).

Source code in teff/channels/webhook.py

async def handle(
    self,
    payload: dict[str, Any],
    *,
    owner: str | None = None,
    headers: Mapping[str, Any] | None = None,
) -> dict:
    """Validate *payload*, run one turn, return the channel response.

    *owner* overrides the configured ``owner:`` spec (the CLI passes the
    resolved value when it wants to override).  The return value matches
    the HTTP channel's shape: ``ok`` plus a turn of ``session_id`` /
    ``waiting`` / ``message`` (the reply, or the interrupt prompt when
    ``waiting``).
    """
    errors = self.validate(payload)
    if errors:
        return {"ok": False, "errors": errors}
    session_id = self.session_id_for(payload)
    result = await self.assistant.run(
        session_id,
        self.message_for(payload),
        owner=owner or self.owner_for(payload, headers),
        max_iterations=self.max_iterations,
    )
    return {"ok": True, **turn_response(result, session_id)}

message_for

message_for(payload)

Render the one-turn message from the payload fields.

Source code in teff/channels/webhook.py

def message_for(self, payload: dict[str, Any]) -> str:
    """Render the one-turn ``message`` from the payload fields."""
    return render_template(self.message_template, payload)

owner_for

owner_for(payload, headers=None)

Resolve the checkpoint owner from the configured owner spec.

payload.<field> reads the body, header.<Name> reads a request header (case-insensitive), fixed:<value> is a constant, and anything else falls back to the spec verbatim (default).

Source code in teff/channels/webhook.py

def owner_for(
    self,
    payload: Mapping[str, Any],
    headers: Mapping[str, Any] | None = None,
) -> str:
    """Resolve the checkpoint owner from the configured ``owner`` spec.

    ``payload.<field>`` reads the body, ``header.<Name>`` reads a
    request header (case-insensitive), ``fixed:<value>`` is a constant,
    and anything else falls back to the spec verbatim (``default``).
    """
    spec = self.owner_spec
    if isinstance(spec, str) and spec.startswith("payload."):
        return str(payload.get(spec[len("payload.") :], "default"))
    if isinstance(spec, str) and spec.startswith("header."):
        name = spec[len("header.") :].lower()
        if headers:
            for key, value in headers.items():
                if str(key).lower() == name and value is not None:
                    return str(value)
        return "default"
    if isinstance(spec, str) and spec.startswith("fixed:"):
        return spec[len("fixed:") :]
    return str(spec)

session_id_for

session_id_for(payload)

Derive the durable session id from the payload.

Uses session_key when configured; otherwise a content hash, so the same payload always resumes the same conversation.

Source code in teff/channels/webhook.py

def session_id_for(self, payload: dict[str, Any]) -> str:
    """Derive the durable session id from the payload.

    Uses ``session_key`` when configured; otherwise a content hash, so
    the same payload always resumes the same conversation.
    """
    if self.session_key is not None:
        value = payload.get(self.session_key)
        if value is not None:
            return str(value)
    raw = str(payload).encode("utf-8", "replace")
    return "wh-" + hashlib.sha1(raw).hexdigest()[:24]

validate

validate(payload)

Return schema errors for payload (empty when valid).

Source code in teff/channels/webhook.py

def validate(self, payload: Any) -> list[str]:
    """Return schema errors for *payload* (empty when valid)."""
    if not self.schema:
        return []
    if not isinstance(payload, dict):
        return ["payload must be an object"]
    return validate_json(payload, self.schema)

HTTPChannel

HTTPChannel(assistant, *, dependencies=None, turn_kwargs=None)

Build an :class:~teff.channels.http.HTTPChannel (needs teff[channels]).

Source code in teff/channels/__init__.py

def HTTPChannel(assistant, *, dependencies=None, turn_kwargs=None):  # noqa: N802
    """Build an :class:`~teff.channels.http.HTTPChannel` (needs ``teff[channels]``)."""
    from teff.channels.http import HTTPChannel as _cls

    return _cls(assistant, dependencies=dependencies, turn_kwargs=turn_kwargs)

build_assistant

build_assistant(path, *, checkpointer=None, max_iterations=80)

Compile path into a durable, interrupt-aware :class:Assistant.

The workflow's checkpoint: block is honored by default (a JSONFileCheckpointer whose path resolves relative to the YAML file); pass checkpointer to override. The state.initial mapping becomes the fresh-session seed, and state.schema reducers apply on every turn.

Parameters:

Name	Type	Description	Default
path	str	Path to the workflow YAML file.	required
checkpointer	Checkpointer | None	Override the checkpointer declared in the file.	None
max_iterations	int	Cap on graph iterations per turn.	80

Returns:

Type	Description
Assistant	A compiled :class:Assistant ready for run()/stream().

Source code in teff/channels/factory.py

def build_assistant(
    path: str,
    *,
    checkpointer: Checkpointer | None = None,
    max_iterations: int = 80,
) -> Assistant:
    """Compile *path* into a durable, interrupt-aware :class:`Assistant`.

    The workflow's ``checkpoint:`` block is honored by default (a
    ``JSONFileCheckpointer`` whose ``path`` resolves relative to the YAML
    file); pass *checkpointer* to override.  The ``state.initial`` mapping
    becomes the fresh-session seed, and ``state.schema`` reducers apply on
    every turn.

    Args:
        path: Path to the workflow YAML file.
        checkpointer: Override the checkpointer declared in the file.
        max_iterations: Cap on graph iterations per turn.

    Returns:
        A compiled :class:`Assistant` ready for ``run()``/``stream()``.
    """
    graph, tools, initial_state, reducers = load_workflow(path)
    if checkpointer is None:
        checkpointer = checkpointer_from_workflow(path)
    from typing import Callable

    make_state: Callable[[], dict] | None = None
    if initial_state:
        seed = dict(initial_state)

        def make_state():
            return dict(seed)

    return Assistant(
        graph,
        tools,
        checkpointer,
        reducers=reducers,
        initial_state=make_state,
        messages_key="messages",
        max_iterations=max_iterations,
    )

build_webhook

build_webhook(assistant, spec)

Build one generic webhook channel from a channels.webhook entry.

Source code in teff/channels/factory.py

def build_webhook(
    assistant: Assistant,
    spec: dict[str, Any],
):
    """Build one generic webhook channel from a ``channels.webhook`` entry."""
    from teff.channels.webhook import WebhookChannel

    return WebhookChannel(assistant, spec)

create_http_app

create_http_app(assistant, *, dependencies=None, turn_kwargs=None)

Build the HTTP/SSE FastAPI app for assistant (needs teff[channels]).

dependencies are FastAPI Depends objects applied to every non-health endpoint; turn_kwargs is a (owner, session_id) -> kwargs factory merged into every Assistant.run/stream call.

Source code in teff/channels/__init__.py

def create_http_app(assistant, *, dependencies=None, turn_kwargs=None):
    """Build the HTTP/SSE FastAPI app for *assistant* (needs ``teff[channels]``).

    ``dependencies`` are FastAPI ``Depends`` objects applied to every
    non-health endpoint; ``turn_kwargs`` is a ``(owner, session_id) -> kwargs``
    factory merged into every ``Assistant.run``/``stream`` call.
    """
    from teff.channels.http import create_http_app as _factory

    return _factory(assistant, dependencies=dependencies, turn_kwargs=turn_kwargs)

create_http_router

create_http_router(assistant, *, dependencies=None, turn_kwargs=None)

Build the HTTP/SSE routes for assistant as a mountable APIRouter.

Use it to embed a channel into an existing app:

from teff.channels import create_http_router
app.include_router(create_http_router(assistant))

Source code in teff/channels/__init__.py

def create_http_router(assistant, *, dependencies=None, turn_kwargs=None):
    """Build the HTTP/SSE routes for *assistant* as a mountable APIRouter.

    Use it to embed a channel into an existing app:

        from teff.channels import create_http_router
        app.include_router(create_http_router(assistant))
    """
    from teff.channels.http import create_http_router as _factory

    return _factory(assistant, dependencies=dependencies, turn_kwargs=turn_kwargs)

load_channels

load_channels(path)

Return the parsed channels: block of path ({} when absent).

The block is read from the raw YAML document, after environment interpolation and include resolution, so ${VAR} references and team/ includes behave like everywhere else in the workflow.

Source code in teff/channels/factory.py

def load_channels(path: str) -> dict:
    """Return the parsed ``channels:`` block of *path* (``{}`` when absent).

    The block is read from the raw YAML document, after environment
    interpolation and include resolution, so ``${VAR}`` references and
    ``team/`` includes behave like everywhere else in the workflow.
    """
    from teff.yaml import _interpolate_env, _load_workflow_document, _resolve_includes

    base_dir = os.path.dirname(os.path.abspath(path))
    data = _interpolate_env(_load_workflow_document(path))
    data = _resolve_includes(data, base_dir)
    channels = data.get("channels", {})
    if channels is None:
        return {}
    if not isinstance(channels, dict):
        raise TypeError("channels: must be a mapping")
    return channels

reply_from_state

reply_from_state(state)

Extract the best-effort assistant reply from a finished state.

Source code in teff/channels/reply.py

def reply_from_state(state: dict | None) -> str:
    """Extract the best-effort assistant reply from a finished *state*."""
    if not state:
        return ""
    for key in _REPLY_KEYS:
        value = state.get(key)
        if isinstance(value, str) and value:
            return value
    messages = state.get("messages")
    if isinstance(messages, list):
        for message in reversed(messages):
            if isinstance(message, dict) and message.get("role") == "assistant":
                content = message.get("content")
                if isinstance(content, str) and content:
                    return content
    return ""

reply_text

reply_text(result)

Return the best-effort assistant reply for result.

"" when the turn is paused (waiting) or produced no text.

Source code in teff/channels/reply.py

def reply_text(result: TurnResult) -> str:
    """Return the best-effort assistant reply for *result*.

    ``""`` when the turn is paused (``waiting``) or produced no text.
    """
    if result.waiting:
        return ""
    if result.reply:
        return result.reply
    return reply_from_state(result.state)

turn_response

turn_response(result, session_id)

Shape one turn into the channel response format.

A paused turn carries the interrupt prompt as message plus the optional key; a completed turn carries the final reply. Used by every channel so HTTP, webhook and Telegram answer identically.

Source code in teff/channels/reply.py

def turn_response(result: TurnResult, session_id: str) -> dict[str, Any]:
    """Shape one turn into the channel response format.

    A paused turn carries the interrupt prompt as ``message`` plus the
    optional ``key``; a completed turn carries the final reply.  Used by
    every channel so HTTP, webhook and Telegram answer identically.
    """
    if result.waiting:
        payload: dict[str, Any] = {
            "session_id": session_id,
            "waiting": True,
            "message": result.prompt or "",
        }
        if result.key:
            payload["key"] = result.key
        return payload
    return {
        "session_id": session_id,
        "waiting": False,
        "message": reply_text(result),
    }