Quickstart

import { AIClientIdentity } from "stream-chat-react/ai-components";

const chat = new AgentChat<LocalMessage>({
  sessionId,
  open: (id) =>
    id ? sessions.resume(id, { tools }) : sessions.create({ tools }),
});
const conversation = new AIConversationController({
  transport: agentChatTransport(chat, {
    adapt: (message, reasoningText) =>
      fromStreamMessage(message, { reasoningText }),
  }),
  actions: { edit: false, regenerate: false },
  approver: {
    userId: "jlahey",
    clientId: AIClientIdentity.installId(),
    decide: async (call, allowed) =>
      (await chat.session()).decide(call.id, { allowed }),
  },
});

import { AIClientIdentity } from "stream-chat-expo/ai-components";

const chat = new AgentChat<LocalMessage>({
  sessionId,
  open: (id) =>
    id ? sessions.resume(id, { tools }) : sessions.create({ tools }),
});
const conversation = new AIConversationController({
  transport: agentChatTransport(chat, {
    adapt: (message, reasoningText) =>
      fromStreamMessage(message, { reasoningText }),
  }),
  actions: { edit: false, regenerate: false },
  approver: {
    userId: "jlahey",
    clientId: AIClientIdentity.installId(),
    decide: async (call, allowed) =>
      (await chat.session()).decide(call.id, { allowed }),
  },
});

28 importsimport androidx.activity.compose.setContent
import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.fillMaxSize
import androidx.compose.runtime.Composable
import androidx.compose.ui.Modifier
import androidx.lifecycle.lifecycleScope
import io.getstream.chat.android.ai.compose.conversation.AIAgentActivity
import io.getstream.chat.android.ai.compose.conversation.AIConversationFeed
import io.getstream.chat.android.ai.compose.conversation.AIConversationRequest
import io.getstream.chat.android.ai.compose.conversation.AIConversationState
import io.getstream.chat.android.ai.compose.conversation.AIConversationTransport
import io.getstream.chat.android.ai.compose.conversation.AIMessage
import io.getstream.chat.android.ai.compose.conversation.AIMessageRole
import io.getstream.chat.android.ai.compose.parts.AIClientIdentity
import io.getstream.chat.android.ai.compose.parts.AIMessagePart
import io.getstream.chat.android.ai.compose.ui.component.AIConversation
import io.getstream.chat.android.ai.compose.ui.component.AIToolApprover
import io.getstream.chat.android.ai.compose.ui.component.ChatComposer
import io.getstream.chat.android.client.channel.state.ChannelState
import io.getstream.chat.android.models.Message
import io.getstream.agents.android.core.AgentEvent
import io.getstream.agents.android.core.AgentSession
import io.getstream.agents.android.core.Conversation
import io.getstream.agents.android.core.channelState
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.launch

// In your activity, once the user is signed in as in step 3.
lifecycleScope.launch {
    val session = startSession()
    // Connects to Stream Chat as the user set above and watches the session's channel.
    val channel = session.channelState(this@MainActivity, lifecycleScope)
    // The user set above, who answers the calls the agent addresses to them.
    val approver = AIToolApprover("jlahey", AIClientIdentity.installId(this@MainActivity), lifecycleScope) { call, allowed ->
        session.decide(call.id, allowed)
    }
    val conversation = AIConversationState(AgentTransport(session, channel, lifecycleScope), lifecycleScope, approver)
    setContent { AgentConversation(conversation, Modifier.fillMaxSize()) }
}

@Composable
fun AgentConversation(conversation: AIConversationState<AgentMessage>, modifier: Modifier = Modifier) {
    Column(modifier) {
        // Each question moves to the top, and its reply streams in below it with its
        // reasoning and tool calls. While the agent answers, the composer shows a Stop button.
        AIConversation(state = conversation, modifier = Modifier.weight(1f))
        ChatComposer(state = conversation)
    }
}

/** Connects the AI components to the session: they show its channel, and ask the session. */
class AgentTransport(
    val session: AgentSession,
    private val channel: StateFlow<ChannelState?>,
    private val scope: CoroutineScope,
) : AIConversationTransport<AgentMessage> {

    override fun connect(feed: AIConversationFeed<AgentMessage>) {
        // The channel's messages, oldest first, as Stream Chat keeps them up to date.
        scope.launch {
            channel.collectLatest { state ->
                state?.messages?.collect { messages ->
                    feed.update(messages.filter { it.deletedAt == null }.map(::AgentMessage))
                }
            }
        }
        // What the agent is doing while its reply shows nothing yet.
        scope.launch {
            session.conversation.collect { conversation ->
                feed.update(
                    activity = when (val state = conversation.state) {
                        Conversation.State.Responding -> AIAgentActivity(AIAgentActivity.State.Thinking)
                        is Conversation.State.Working -> AIAgentActivity(
                            AIAgentActivity.State.CheckingExternalSources,
                            "Working on " + state.skills.joinToString(" and "),
                        )
                        else -> null
                    },
                )
            }
        }
        // A reply the model couldn't write leaves nothing in the channel: only the session says so.
        scope.launch {
            session.events().collect { event ->
                if (event.kind == AgentEvent.Kind.Error && event.string("context") in setOf("llm", "sts")) {
                    feed.update(activity = AIAgentActivity(AIAgentActivity.State.Error))
                }
            }
        }
    }

    // The agent does not answer a message sent to the channel: ask the session.
    override suspend fun send(request: AIConversationRequest<AgentMessage>) {
        session.responses.create(request.text)
    }

    override suspend fun stop() {
        session.interrupt()
    }
}

/** A message of the channel, as the AI components show it. */
data class AgentMessage(val message: Message) : AIMessage {
    override val id: String = message.id
    // The agent's replies carry `ai_generated: true`.
    override val role: AIMessageRole =
        if (message.extraData["ai_generated"] == true) AIMessageRole.Assistant else AIMessageRole.User
    override val text: String = message.text
    override val isGenerating: Boolean = message.extraData["generating"] == true
    // The reply's reasoning and tool calls, which the router writes as its attachments. Stream Chat
    // lifts `name` out of an attachment's fields, and a tool call's step needs it back.
    override val parts: List<AIMessagePart> = AIMessagePart.parts(
        message.attachments.map { it.type.orEmpty() to it.extraData + ("name" to it.name) },
    )
}

import Combine
import StreamChat
import StreamChatAI
import SwiftUI
import StreamAgents

/// Opens a session with the agent, and shows it once its channel is open.
struct AgentScreen: View {
    @State private var transport: AgentTransport?
    @State private var failure: Error?

    var body: some View {
        if let transport {
            AgentConversation(transport: transport)
        } else if let failure {
            // Such as a CLI token that expired after its 2 hours: mint a new one.
            VStack(spacing: 16) {
                Text(failure.localizedDescription)
                Button("Try again") { self.failure = nil }
            }
            .padding()
        } else {
            ProgressView()
                .task {
                    do {
                        transport = try await AgentTransport(startSession())
                    } catch {
                        failure = error
                    }
                }
        }
    }
}

struct AgentConversation: View {
    @StateObject private var conversation: AIConversation<AgentMessage>

    init(transport: AgentTransport) {
        // Rating only: regenerating or editing would ask the agent again, and show the question twice.
        _conversation = StateObject(
            wrappedValue: AIConversation(transport: transport, approver: transport.approver, actions: [.rate])
        )
    }

    var body: some View {
        // Each question moves to the top, and its reply streams in below it with its
        // reasoning and tool calls. While the agent writes, the composer shows a Stop button.
        AIConversationView(conversation: conversation)
            .safeAreaInset(edge: .bottom) {
                AIComposerView(conversation: conversation)
            }
    }
}

/// Connects the AI components to the session: they show its channel, and ask the session.
@MainActor
final class AgentTransport: AIConversationTransport {
    let session: AgentSession
    private let channel: ChatChannelController
    private var subscription: AnyCancellable?

    init(_ session: AgentSession) async throws {
        self.session = session
        // `session.chat()` connects to Stream Chat as the user set above and opens the session's channel.
        channel = try await session.chat()
        // The session's socket, which stops a reply and runs the tools this phone offers.
        await session.start()
    }

    func connect(_ feed: AIConversationFeed<AgentMessage>) {
        // The channel lists its messages newest first, the conversation oldest first.
        let publish = { [channel] in
            feed.update(messages: channel.messages.reversed().map(AgentMessage.init))
        }
        subscription = channel.messagesChangesPublisher.sink { _ in publish() }
        channel.synchronize { _ in publish() }
    }

    // The agent does not answer a message sent to the channel: ask the session.
    func send(_ request: AIConversationRequest<AgentMessage>) async throws {
        _ = try await session.responses.create(request.text)
    }

    func stop() async throws {
        try await session.interrupt()
    }

    /// Answers the tools that ask before they run.
    var approver: AIToolApprover {
        AIToolApprover(userID: agents.backend.user?.id ?? "", clientID: AIClientIdentity.installID) { call, allowed in
            try await self.session.decide(call.id, allowed: allowed)
        }
    }
}

/// A message of the channel, as the AI components show it.
struct AgentMessage: AIMessage {
    let message: ChatMessage

    var id: MessageId { message.id }
    // The agent's replies carry `ai_generated: true`.
    var role: AIMessageRole {
        message.extraData["ai_generated"]?.boolValue == true ? .assistant : .user
    }
    var text: String { message.text }
    var isGenerating: Bool { message.extraData["generating"]?.boolValue == true }
    // The reply's reasoning and tool calls, which the router writes as its attachments.
    var parts: [AIMessagePart] {
        AIMessagePart.parts(from: message.allAttachments.map { ($0.type.rawValue, $0.payload) })
    }
}

import 'dart:async';

import 'package:flutter/material.dart';
import 'package:stream_chat/stream_chat.dart' as stream;
import 'package:stream_chat_flutter_ai/stream_chat_flutter_ai.dart';
import 'package:stream_agents_chat/stream_agents_chat.dart';
import 'package:stream_agents_core/stream_agents_core.dart';

// Signs the user in once, as in step 3, and shows the agent.
void main() => runApp(
  MaterialApp(home: Scaffold(body: SafeArea(child: AgentScreen(agents: signIn())))),
);

/// Opens a session with the agent, and shows it once its channel is open.
class AgentScreen extends StatefulWidget {
  const AgentScreen({super.key, required this.agents});

  final StreamAgents agents;

  @override
  State<AgentScreen> createState() => _AgentScreenState();
}

class _AgentScreenState extends State<AgentScreen> {
  late final Future<AgentTransport> _transport = startSession(
    widget.agents,
  ).then(AgentTransport.open);

  @override
  Widget build(BuildContext context) => FutureBuilder(
    future: _transport,
    builder: (context, snapshot) {
      if (snapshot.data case final transport?) return AgentConversation(transport: transport);
      if (snapshot.hasError) return Center(child: Text("The conversation couldn't open: ${snapshot.error}"));
      return const Center(child: CircularProgressIndicator());
    },
  );
}

class AgentConversation extends StatefulWidget {
  const AgentConversation({super.key, required this.transport});

  final AgentTransport transport;

  @override
  State<AgentConversation> createState() => _AgentConversationState();
}

class _AgentConversationState extends State<AgentConversation> {
  // Made once and disposed with the screen, so an answer on its way survives a rebuild.
  late final _approver = widget.transport.approver();
  late final _conversation = AIConversation(
    transport: widget.transport,
    approver: _approver,
    // The router keeps the conversation in Stream Chat and can't rewind it.
    actions: AIConversationActions.rate,
  );

  @override
  void dispose() {
    _conversation.dispose();
    _approver.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) => Column(
    children: [
      Expanded(
        // Each question moves to the top, and its reply streams in below it with its
        // reasoning and tool calls. While the agent writes, the composer shows a Stop button.
        child: AIConversationView.conversation(conversation: _conversation),
      ),
      ChatComposer.conversation(conversation: _conversation, enableSpeechToText: true),
    ],
  );
}

/// Connects the AI components to the session: they show its channel, and ask the session.
class AgentTransport with AIConversationTransport<AgentMessage> {
  AgentTransport._(this.session, this._channel);

  static Future<AgentTransport> open(AgentSession session) async {
    // `session.chat()` connects to Stream Chat as the user signed in above and opens the session's channel.
    final channel = await session.chat();
    await channel.watch();
    return AgentTransport._(session, channel);
  }

  final AgentSession session;
  final stream.Channel _channel;
  final _subscriptions = <StreamSubscription<Object?>>[];

  // Whether the latest reply failed, until the next one starts: the session goes on saying
  // the agent responds.
  var _failed = false;

  @override
  void connect(AIConversationFeed<AgentMessage> feed) {
    _subscriptions
      // The channel's messages, oldest first, as Stream Chat keeps them up to date.
      ..add(
        _channel.state!.messagesStream.listen((messages) {
          feed.updateMessages([
            for (final message in messages)
              if (message.deletedAt == null) AgentMessage(message),
          ]);
        }),
      )
      // What the agent is doing while its reply shows nothing yet.
      ..add(
        session.conversation.stream.listen((conversation) {
          if (!_failed) feed.updateActivity(_activityOf(conversation.state));
        }),
      )
      // A reply the model couldn't write leaves nothing in the channel: only the session says so.
      ..add(
        session.events().listen((event) {
          if (event.kind == AgentEventKind.responding) {
            _failed = false;
            feed.updateActivity(_activityOf(session.state));
          } else if (event.kind == AgentEventKind.error && const {'llm', 'sts'}.contains(event['context'])) {
            _failed = true;
            feed.updateActivity(const AIAgentActivity(state: AIAgentActivityState.error));
          }
        }),
      );
  }

  static AIAgentActivity? _activityOf(ConversationState state) => switch (state) {
    Responding() => const AIAgentActivity(state: AIAgentActivityState.thinking),
    Working(:final skills) => AIAgentActivity(
      state: AIAgentActivityState.checkingExternalSources,
      text: 'Working on ${skills.join(' and ')}',
    ),
    _ => null,
  };

  // The agent does not answer a message sent to the channel: ask the session.
  @override
  Future<void> send(AIConversationRequest<AgentMessage> request) async {
    await session.responses.create(request.text);
  }

  @override
  Future<void> stop() async => session.interrupt();

  /// Answers the tools that ask before they run.
  AIToolApprover approver() => AIToolApprover(
    identity: AIClientIdentity(userId: session.backend.userId),
    decide: (call, decision) async {
      // The session ignores an answer to a call it isn't waiting on, such as one asked before
      // the app was relaunched, so say so rather than show the answer as sent.
      if (!session.approvals.value.any((request) => request.id == call.id)) {
        throw StateError('This question is no longer waiting for an answer.');
      }
      session.decide(call.id, allowed: decision == AIToolApprovalDecision.allowed);
    },
  );

  /// Stops following the channel and the session, as the conversation is disposed.
  @override
  void disconnect() {
    for (final subscription in _subscriptions) {
      unawaited(subscription.cancel());
    }
    _subscriptions.clear();
  }
}

/// A message of the channel, as the AI components show it.
class AgentMessage with AIMessage {
  // The reply's reasoning and tool calls, which the router writes as its attachments.
  AgentMessage(this.message)
    : parts = AIMessagePart.parseAll([
        for (final attachment in message.attachments)
          if (attachment.rawType case final type?)
            (type: type, payload: {'id': attachment.id, ...attachment.extraData}),
      ]);

  final stream.Message message;

  @override
  String get id => message.id;

  // The agent's replies carry `ai_generated: true`.
  @override
  AIMessageRole get role =>
      message.extraData['ai_generated'] == true ? AIMessageRole.assistant : AIMessageRole.user;

  @override
  String get text => message.text ?? '';

  @override
  bool get isGenerating => message.extraData['generating'] == true;

  @override
  final List<AIMessagePart> parts;
}