Gateway Threads

Open a native platform thread/topic under a channel — capability-gated, never raises

An agent opens a native thread (Telegram forum topic, Discord thread, Slack thread) under a channel and posts a follow-up into it — falling back to the parent channel when the platform can’t thread.

Quick Start

1. Open a thread and post into it

   from praisonaiagents import Agent
   
   agent = Agent(
       name="assistant",
       instructions="Group replies by topic when the channel supports threads.",
   )
   
   result = agent.gateway.create_thread(target="telegram:123456", name="daily-standup")
   if result.ok:
       agent.gateway.send(target=f"{result.target}:{result.thread_id}", text="Standup starting…")
   else:
       # unsupported | failed | no_route — fall back to the parent channel
       agent.gateway.send(target="telegram:123456", text="Standup starting…")

2. Read the typed outcome

   from praisonaiagents.gateway import ThreadResult
   
   result: ThreadResult = agent.gateway.create_thread(target="slack:C123", name="research")
   
   result.status      # "ok" | "unsupported" | "failed" | "no_route"
   result.ok          # True only when status == "ok"
   result.thread_id   # the new thread id (set on "ok", "" otherwise)
   result.target      # the resolved parent target

How It Works

create_thread resolves the target, checks the channel’s supports_threads capability, and dispatches to the adapter — returning a typed outcome instead of raising.

Status Decision Guide

Each call resolves to exactly one ThreadStatus — map it to the right caller action.

Configuration Options

ThreadResult (from praisonaiagents.gateway) is the typed return of every create_thread call.

ThreadStatus is a string literal — one of the four values below.

The capability flag lives on the adapter’s PlatformCapabilities.

Fallback Pattern

Try the thread, and route to the parent channel on any non-ok outcome.

from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="Scope subtasks into threads when possible.")

parent = "slack:C123"
result = agent.gateway.create_thread(target=parent, name="research")

# Compose the routed target only when a thread was actually opened
destination = f"{result.target}:{result.thread_id}" if result.ok else parent
agent.gateway.send(target=destination, text="Kicking off the research subtask…")

Authoring a Custom Adapter

A custom OutboundMessengerProtocol implementation declares the capability and returns a ThreadResult.

from praisonaiagents.gateway import ThreadResult
from praisonaiagents.bots.protocols import PlatformCapabilities


class MyAdapter:
    @property
    def capabilities(self) -> PlatformCapabilities:
        return PlatformCapabilities(supports_threads=True)

    async def create_thread(self, target: str, name: str) -> ThreadResult:
        thread_id = await self._open_native_topic(target, name)
        if not thread_id:
            return ThreadResult(status="failed", target=target)
        return ThreadResult(status="ok", target=target, thread_id=str(thread_id))

Per-Platform Support

Support tracks each adapter’s supports_threads capability.

Common Patterns

Scope a multi-agent subtask into its own thread so it doesn’t flood the shared channel.

result = agent.gateway.create_thread(target="discord:987654", name="deploy-run")
target = f"{result.target}:{result.thread_id}" if result.ok else "discord:987654"
agent.gateway.send(target=target, text="Deploy started — updates will land here.")

Branch parallel workflow steps into separate threads, each falling back independently.

parent = "telegram:123456"
for name in ("build", "test", "ship"):
    r = agent.gateway.create_thread(target=parent, name=name)
    agent.gateway.send(
        target=f"{r.target}:{r.thread_id}" if r.ok else parent,
        text=f"{name} stage running…",
    )

Best Practices

Related