feat(think): extract transport from Think into a separate "host" DO

.changeset/quiet-clouds-ingest.md

@@ -0,0 +1,7 @@
+---
+"@cloudflare/think": minor
+---
+
+Add `ingest()` for host-owned transports driving Think agents over Workers RPC as a byte-streaming NDJSON contract, and allow policy-only non-messenger channels (omit `kind` — it is now optional and deprecated).
+
+Deprecates the Think-owned transport surface — `getMessengers()`, `messengerChannel()`, and the `kind`/`ingress`/`capabilities`/`conversation`/`delivery` channel fields — in favour of user-owned hosts driving agents via `ingest()`; existing apps keep working unchanged.

docs/think/messengers.md

@@ -1,5 +1,11 @@
 # Think Messengers
 
+> **Deprecated.** The Think-owned messenger runtime is being superseded by
+> user-owned hosts: run the Chat SDK in your worker (with its own state DO)
+> and drive agents via [`ingest()`](./channels.md) over Workers RPC — see
+> `examples/channel-host-telegram`. Existing messenger apps keep working, but
+> new integrations should use the host pattern.
+
 Use messengers when a Think agent should receive and reply to Chat SDK webhooks
 directly. Think owns the webhook route, durable reply fiber, conversation
 routing, and streamed delivery back to the provider.

examples/channel-host-telegram/.dev.vars.example

@@ -0,0 +1,4 @@
+# Copy to .dev.vars (local) and `wrangler secret put` each for deploys.
+TELEGRAM_BOT_TOKEN=123456:ABC-your-botfather-token
+TELEGRAM_WEBHOOK_SECRET_TOKEN=any-random-string-you-choose
+TELEGRAM_BOT_USERNAME=your_bot_username

examples/channel-host-telegram/README.md

@@ -0,0 +1,77 @@
+# channel-host-telegram
+
+A real messenger built on the transport outside, agent as callee pattern: the
+Chat SDK runs in a Worker-owned host, and Think agents are callees invoked over
+Workers RPC.
+
+| Role                             | Where                                                                                                                                               | Owns                                                                                                                                 |
+| -------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
+| Client                           | the Telegram app                                                                                                                                    | nothing you control: raw updates, including slash commands, hit your webhook.                                                        |
+| Host (worker in `src/server.ts`) | `Chat` from the Chat SDK, state in its own `ChatStateDO` (via [`chat-state-cloudflare-do`](https://github.com/dcartertwo/chat-state-cloudflare-do)) | the Telegram connection, webhook verification, threading, dedupe, locks, the command surface, and which agent handles a thread.      |
+| Agent (`HostedAgent`)            | a Think Durable Object, one instance per Telegram thread                                                                                            | the model turn and per-thread transcript. Its public surface is only `ingest()` over RPC plus `configureChannels()` behavior policy. |
+
+> Note: Chat SDK state uses the community
+> [`chat-state-cloudflare-do`](https://chat-sdk.dev/adapters/community/cloudflare-do)
+> adapter. It should be vendored into this repo before the host pattern is
+> promoted beyond experimental.
+
+Key differences from the built-in `getMessengers()` path:
+
+- The Chat SDK instance and adapter options are ordinary app code in the host.
+- The Worker only serves `/webhooks/telegram` and `/setup/telegram`.
+- The agent has no public transport route, static routing manifest, or webhook
+  handler.
+- Trade-off you now own: reply durability. The built-in messenger runtime wraps
+  delivery in recovery fibers; this host posts once per turn and relies on
+  Telegram webhook retries plus Chat SDK dedupe for at-least-once ingress.
+
+The agent surface is literally two things:
+
+```typescript
+async ingest(input: {
+  channelId: "telegram";
+  text: string;
+}): Promise<ReadableStream<Uint8Array>>;
+
+configureChannels() {
+  return {
+    telegram: {
+      instructions: "Telegram-specific behavior policy"
+    }
+  };
+}
+```
+
+The returned stream is UTF-8 NDJSON with `delta`, `done`, and `error` frames.
+This host decodes `delta` frames and passes an `AsyncIterable<string>` to
+`thread.post(...)`, letting the Chat SDK stream by posting and editing the
+Telegram reply.
+
+## Setup
+
+1. Create a bot with [@BotFather](https://t.me/BotFather), note the token and
+   the bot's username.
+2. `cp .dev.vars.example .dev.vars` and fill in the three values
+   (`TELEGRAM_WEBHOOK_SECRET_TOKEN` is any random string you choose).
+3. Deploy and register the webhook. Telegram needs a public HTTPS URL, so local
+   `wrangler dev` cannot receive real messages:
+
+```sh
+pnpm deploy
+npx wrangler secret bulk <(node -e '
+  const fs=require("fs");const o={};
+  for(const l of fs.readFileSync(".dev.vars","utf8").split("\n")){
+    const m=l.match(/^([A-Z_]+)=(.*)$/); if(m) o[m[1]]=m[2];
+  } console.log(JSON.stringify(o))')
+curl https://channel-host-telegram.<your-subdomain>.workers.dev/setup/telegram
+```
+
+4. DM the bot on Telegram:
+
+```text
+/help            -> answered by the host, no agent DO, no model turn
+/whoami          -> answered by the host, shows the thread's agent instance
+anything else    -> Workers AI turn in the thread's Think agent
+```
+
+`npx wrangler tail` while you chat to watch the host and agent split in the logs.

examples/channel-host-telegram/env.d.ts

@@ -0,0 +1,11 @@
+// Generated by `pnpm types`; kept minimal by hand for the demo.
+interface Env {
+  AI: Ai;
+  HostedAgent: DurableObjectNamespace<import("./src/server").HostedAgent>;
+  CHAT_STATE: DurableObjectNamespace<
+    import("chat-state-cloudflare-do").ChatStateDO
+  >;
+  TELEGRAM_BOT_TOKEN: string;
+  TELEGRAM_WEBHOOK_SECRET_TOKEN: string;
+  TELEGRAM_BOT_USERNAME?: string;
+}

examples/channel-host-telegram/package.json

@@ -0,0 +1,27 @@
+{
+  "name": "@cloudflare/agents-channel-host-telegram",
+  "description": "Channel-host inversion demo: a worker-owned Chat SDK Telegram bot (state in its own DO) driving Think agents as callees",
+  "private": true,
+  "type": "module",
+  "version": "0.0.0",
+  "scripts": {
+    "start": "wrangler dev",
+    "deploy": "wrangler deploy",
+    "typecheck": "tsc --noEmit",
+    "types": "wrangler types env.d.ts --include-runtime false"
+  },
+  "dependencies": {
+    "@chat-adapter/telegram": "^4.31.0",
+    "@cloudflare/think": "*",
+    "agents": "*",
+    "ai": "^6.0.202",
+    "chat": "^4.31.0",
+    "chat-state-cloudflare-do": "^0.2.0"
+  },
+  "devDependencies": {
+    "@cloudflare/workers-types": "^4.20260629.1",
+    "@types/node": "^26.0.1",
+    "typescript": "^6.0.3",
+    "wrangler": "^4.105.0"
+  }
+}

examples/channel-host-telegram/src/server.ts

@@ -0,0 +1,158 @@
+import { createTelegramAdapter } from "@chat-adapter/telegram";
+import type { ThinkChannels } from "@cloudflare/think";
+import { decodeIngestStream, Think } from "@cloudflare/think";
+import { getAgentByName } from "agents";
+import { Chat } from "chat";
+import { createCloudflareState } from "chat-state-cloudflare-do";
+
+// The Chat SDK's state lives in its OWN Durable Object — no Think involved.
+export { ChatStateDO } from "chat-state-cloudflare-do";
+
+const WEBHOOK_PATH = "/webhooks/telegram";
+
+// ─────────────────────────────────────────────────────────────────────────────
+// The AGENT: a Durable Object with zero transport code. It does not know
+// Telegram exists. One instance per Telegram thread gets its own persistent
+// transcript. The host reaches it through Workers RPC.
+// ─────────────────────────────────────────────────────────────────────────────
+export class HostedAgent extends Think<Env> {
+  getModel() {
+    return "@cf/moonshotai/kimi-k2.7-code";
+  }
+
+  configureChannels(): ThinkChannels {
+    return {
+      telegram: {
+        instructions:
+          "You are replying inside a Telegram chat. Be concise and direct. Plain text only, no markdown tables or headers."
+      }
+    };
+  }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// The HOST: a worker-owned Chat SDK bot. It owns the Telegram connection,
+// webhook verification, threading/dedupe state (in ChatStateDO), and the
+// command surface. The agent is a callee it invokes for real messages.
+// ─────────────────────────────────────────────────────────────────────────────
+
+function safeAgentName(threadId: string): string {
+  return threadId.replace(/[^A-Za-z0-9_-]+/g, "-") || "default";
+}
+
+/** Ask the agent for a turn over native Workers RPC. */
+async function* askAgent(
+  env: Env,
+  threadId: string,
+  text: string
+): AsyncIterable<string> {
+  const agent = await getAgentByName(env.HostedAgent, safeAgentName(threadId));
+  const stream = await agent.ingest({
+    channelId: "telegram",
+    text
+  });
+  // Buffered alternative: const reply = await collectIngestReply(stream);
+  for await (const event of decodeIngestStream(stream)) {
+    if (event.type === "delta") yield event.text;
+    if (event.type === "error") throw new Error(event.message);
+  }
+}
+
+function createBot(env: Env) {
+  const userName = env.TELEGRAM_BOT_USERNAME ?? "channel_host_demo_bot";
+  const chat = new Chat({
+    adapters: {
+      telegram: createTelegramAdapter({
+        botToken: env.TELEGRAM_BOT_TOKEN,
+        mode: "webhook",
+        secretToken: env.TELEGRAM_WEBHOOK_SECRET_TOKEN,
+        userName
+      })
+    },
+    state: createCloudflareState({ namespace: env.CHAT_STATE }),
+    userName
+  });
+
+  // Host-owned commands. The adapter parses `/commands` out of the update
+  // stream and routes them here — they never become messages, never wake the
+  // agent DO, never run a model turn.
+  chat.onSlashCommand(async (event) => {
+    switch (event.command) {
+      case "/help":
+        await event.channel.post(
+          "Commands: /help, /whoami. Anything else goes to the AI agent."
+        );
+        return;
+      case "/whoami":
+        await event.channel.post(
+          `channel "${event.channel.id}" -> dedicated agent instance "${safeAgentName(event.channel.id)}"`
+        );
+        return;
+      default:
+        await event.channel.post(
+          `Unknown command ${event.command}. The agent never sees slash commands.`
+        );
+    }
+  });
+
+  // Real messages stream from the agent straight into Telegram (post+edit).
+  const respond = async (thread: { id: string; post: Poster }, text?: string) =>
+    text?.trim() && thread.post(askAgent(env, thread.id, text.trim()));
+
+  chat.onDirectMessage(async (thread, message) => {
+    await respond(thread, message.text);
+  });
+  chat.onNewMention(async (thread, message) => {
+    await thread.subscribe();
+    await respond(thread, message.text);
+  });
+  chat.onSubscribedMessage(async (thread, message) => {
+    await respond(thread, message.text);
+  });
+
+  return chat;
+}
+
+type Poster = (text: string | AsyncIterable<string>) => Promise<unknown>;
+
+/** One-time (idempotent) webhook registration: GET /setup/telegram */
+async function setupWebhook(request: Request, env: Env): Promise<Response> {
+  const origin = new URL(request.url).origin;
+  const webhookUrl = `${origin}${WEBHOOK_PATH}`;
+  const response = await fetch(
+    `https://api.telegram.org/bot${env.TELEGRAM_BOT_TOKEN}/setWebhook`,
+    {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        url: webhookUrl,
+        secret_token: env.TELEGRAM_WEBHOOK_SECRET_TOKEN,
+        allowed_updates: ["message"]
+      })
+    }
+  );
+  const result = (await response.json()) as {
+    ok: boolean;
+    description?: string;
+  };
+  return Response.json({ webhookUrl, ...result });
+}
+
+export default {
+  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
+    const url = new URL(request.url);
+    if (url.pathname === WEBHOOK_PATH && request.method === "POST") {
+      // The adapter verifies Telegram's secret token itself (secretToken in
+      // its config), so the host adds no verification code of its own.
+      const bot = createBot(env);
+      return bot.webhooks.telegram(request, {
+        waitUntil: (promise) => ctx.waitUntil(promise)
+      });
+    }
+    if (url.pathname === "/setup/telegram") {
+      return setupWebhook(request, env);
+    }
+    // The agent has no public transport route. The host is the only door.
+    return new Response("Not found", { status: 404 });
+  }
+} satisfies ExportedHandler<Env>;

examples/channel-host-telegram/tsconfig.json

@@ -0,0 +1,4 @@
+{
+  "extends": "agents/tsconfig",
+  "include": ["src", "env.d.ts"]
+}

examples/channel-host-telegram/wrangler.jsonc

@@ -0,0 +1,31 @@
+{
+  "$schema": "./node_modules/wrangler/config-schema.json",
+  "name": "channel-host-telegram",
+  "main": "src/server.ts",
+  "compatibility_date": "2026-06-11",
+  "compatibility_flags": ["nodejs_compat"],
+  "ai": { "binding": "AI", "remote": true },
+  "durable_objects": {
+    "bindings": [
+      {
+        "name": "HostedAgent",
+        "class_name": "HostedAgent"
+      },
+      {
+        "name": "CHAT_STATE",
+        "class_name": "ChatStateDO"
+      }
+    ]
+  },
+  "migrations": [
+    {
+      "tag": "v1",
+      "new_sqlite_classes": ["HostedAgent", "ChatStateDO"]
+    }
+  ],
+  "observability": {
+    "logs": {
+      "enabled": true
+    }
+  }
+}