Channels

Example

package com.example.tnsai.docs;

import com.tnsai.agents.Agent;
import com.tnsai.channels.api.UnifiedResponse;
import com.tnsai.channels.registry.ChannelRegistry;
import com.tnsai.channels.routing.MessageRouter;
import com.tnsai.channels.slack.SlackChannel;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;

/** Compilation contract for connect/channels.md. */
public final class ChannelsExample {

    private ChannelsExample() {
    }

    static MessageRouter telegram(Supplier<Agent> newAgent) {
        ChannelRegistry registry = new ChannelRegistry();
        registry.discoverAdapters();                     // every adapter listed in META-INF/services

        Map<String, Agent> agents = new ConcurrentHashMap<>();
        MessageRouter router = new MessageRouter(registry);
        router.setMessageHandler((message, reply) -> {
            if (message.text() == null) {
                return;                                  // media-only message
            }
            // One agent per conversation, so chats do not share history.
            Agent agent = agents.computeIfAbsent(message.sessionKey(), key -> newAgent.get());
            reply.send(UnifiedResponse.builder()
                .channelId(message.channelId())
                .conversationId(message.conversationId())
                .text(agent.chat(message.text()))
                .build());
        });

        registry.startAdapter("telegram", router);       // reads TELEGRAM_BOT_TOKEN
        return router;
    }

    static void slack(MessageRouter router, ChannelRegistry registry) {
        // Without discoverAdapters(); a second adapter with the same id is ignored.
        registry.register(new SlackChannel());           // reads SLACK_BOT_TOKEN and SLACK_APP_TOKEN
        registry.startAdapter("slack", router);
        if (!registry.get("slack").orElseThrow().isRunning()) {
            throw new IllegalStateException("Slack did not start; see the router's error log");
        }
    }

    static int reap(MessageRouter router) {
        return router.getSessions().reapIdle(Duration.ofHours(1));   // nothing reaps on its own
    }
}

Adapters

Each adapter except telegram and cli also takes a config record (SlackChannelConfig and so on) in its constructor. Without one, it reads the environment when it starts.

• Discord's default intents include message content, which you must turn on for the bot in the Discord developer portal.
• The WhatsApp webhook listens on 127.0.0.1:8443/webhook by default (WHATSAPP_WEBHOOK_BIND_ADDRESS, WHATSAPP_WEBHOOK_PORT, WHATSAPP_WEBHOOK_PATH). Put a public HTTPS proxy or tunnel in front of it. It rejects events whose X-Hub-Signature-256 does not match WHATSAPP_APP_SECRET.
• The email adapter only answers senders on EMAIL_SENDER_ALLOWLIST, a comma-separated list. Write the addresses in lowercase, because incoming addresses are lowercased before the check. With an empty list it answers no one.
• The Telegram, Slack, Discord and WhatsApp adapters answer anyone who can message the bot.

Start adapters

discoverAdapters() registers every adapter listed in META-INF/services/com.tnsai.channels.api.ChannelAdapter, which is all six. Start the ones you configured with startAdapter(id, router). startAll(router) also starts the rest: the CLI adapter takes over standard input, and each unconfigured adapter reports an error. register(adapter) adds one adapter yourself. A second adapter with an id already registered is ignored.

When an adapter has no configuration, it reports a fatal ChannelError to the router and returns. The router logs the error, but getState(id) still says RUNNING, because the adapter did not throw. Check adapter.isRunning() instead. getState returns FAILED only when start throws.

Handle messages

MessageRouter calls your handler on the thread that delivered the message, so a slow agent call holds up that adapter. The reply goes to the adapter the message came from. The response's channelId is not used for routing. Without a handler, the router logs an error and drops the message.

message.sessionKey() is channelId:conversationId. A single agent keeps one chat history, so give each conversation its own agent, as the example does. getSessions() records when each conversation was last active. Nothing removes old entries until you call reapIdle(maxIdle).

sendProactive(channelId, conversationId, text) sends a message that is not a reply.

Messages

Telegram uses showTyping and replyToMessageId. Slack and Discord use replyToMessageId, which on Slack is the thread timestamp. No adapter reads streaming or splitLongMessages.

Stream replies

StreamingChannelAdapter adds sendChunk(UnifiedChunk) for text that arrives in pieces. Only CliChannel implements it, and MessageRouter never calls it, so your handler sends the chunks to the adapter itself. UnifiedChunk.text(conversationId, delta) makes a piece and UnifiedChunk.done(conversationId) ends the reply. After a streamed reply, CliChannel skips the next send(...), because the text is already on screen.

Write an adapter

1. Implement ChannelAdapter: id(), displayName(), capabilities(), start(context), stop(), send(response) and isRunning(). Call context.onMessage(...) for each incoming message, and context.onError(...) with a TRANSIENT, DEGRADED or FATAL severity when something fails.
2. List the adapter in META-INF/services/com.tnsai.channels.api.ChannelAdapter. This is the file discoverAdapters() reads. The adapter needs a public no-argument constructor.
3. Optionally implement ChannelAdapterFactory and list it in META-INF/services/com.tnsai.channels.api.ChannelAdapterFactory. ChannelDiscovery.discoverFactories() returns the factories whose isAvailable() is true. The registry does not use factories.

@Channel(id, displayName) documents an adapter class. Nothing reads it, so it does not register anything.

Related

• Connect
• Create an agent
• Server