From 4440f7cae349ede481d7dd1774e0b222707e19a0 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Wed, 8 Jul 2026 21:28:05 +0000 Subject: [PATCH] feat(agents): send console server info to CLI --- .changeset/console-server-info.md | 5 +++++ agents/src/cli.ts | 4 ++++ agents/src/console.ts | 11 +++++++++- agents/src/voice/remote_session.ts | 32 +++++++++++++++++++++++++++--- 4 files changed, 48 insertions(+), 4 deletions(-) create mode 100644 .changeset/console-server-info.md diff --git a/.changeset/console-server-info.md b/.changeset/console-server-info.md new file mode 100644 index 000000000..044985f4d --- /dev/null +++ b/.changeset/console-server-info.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': minor +--- + +Send agent server info to the LiveKit CLI host when starting a console session. diff --git a/agents/src/cli.ts b/agents/src/cli.ts index 00b2c7813..398c3de2e 100644 --- a/agents/src/cli.ts +++ b/agents/src/cli.ts @@ -212,11 +212,15 @@ export const runApp = (opts: ServerOptions) => { .action((...[, command]) => { const commandOptions = command.opts(); opts.logLevel = commandOptions.logLevel; + const globalOptions = program.optsWithGlobals(); + opts.wsURL = globalOptions.url || opts.wsURL; initializeLogger({ pretty: true, level: opts.logLevel }); process.env.LIVEKIT_DEV_MODE = '1'; runConsole({ agentPath: opts.agent, connectAddr: commandOptions.connectAddr, + agentName: opts.agentName, + wsURL: opts.wsURL, record: commandOptions.record === true, }).catch((error) => { log().fatal(`console mode failed: ${formatErrorMessage(error)}`); diff --git a/agents/src/console.ts b/agents/src/console.ts index 8c18af77c..fdfc40bed 100644 --- a/agents/src/console.ts +++ b/agents/src/console.ts @@ -55,10 +55,14 @@ async function loadAgent(agentPath: string): Promise { export async function runConsole({ agentPath, connectAddr, + agentName, + wsURL, record, }: { agentPath: string; connectAddr: string; + agentName: string; + wsURL: string; record: boolean; }): Promise { const logger = log(); @@ -77,7 +81,12 @@ export async function runConsole({ const agent = await loadAgent(agentPath); const prewarm = agent.prewarm ?? defaultInitializeProcessFunc; - const transport = new TcpSessionTransport(host, port); + const transport = new TcpSessionTransport(host, port, { + serverInfo: { + agentName, + url: wsURL, + }, + }); const audioInput = new TcpAudioInput(); const audioOutput = new TcpAudioOutput(transport); diff --git a/agents/src/voice/remote_session.ts b/agents/src/voice/remote_session.ts index ae2af7bbe..9dffd58b0 100644 --- a/agents/src/voice/remote_session.ts +++ b/agents/src/voice/remote_session.ts @@ -2,7 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import { Duration, Timestamp } from '@bufbuild/protobuf'; -import { AgentSession as pb } from '@livekit/protocol'; +import { AgentDev, AgentSession as pb } from '@livekit/protocol'; import type { ByteStreamReader, Room, TextStreamInfo } from '@livekit/rtc-node'; import { ThrowsPromise } from '@livekit/throws-transformer/throws'; import type { TypedEventEmitter } from '@livekit/typed-emitter'; @@ -241,6 +241,13 @@ const TCP_HEADER_SIZE = 4; const TCP_MAX_MESSAGE_SIZE = 1 << 20; // 1 MiB const TCP_DRAIN_THRESHOLD = 64 * 1024; // 64 KiB +type TcpSessionTransportOptions = { + serverInfo?: { + agentName: string; + url: string; + }; +}; + /** * Resolve once the socket's write buffer drains, or when the socket closes or * errors. Unlike a bare `once('drain')`, this can't hang forever if the peer @@ -268,13 +275,19 @@ function waitForDrain(socket: net.Socket): Promise { export class TcpSessionTransport extends SessionTransport { private readonly host: string; private readonly port: number; + private readonly options: TcpSessionTransportOptions; private socket: net.Socket | null = null; private closed = false; - constructor(host: string, port: number) { + constructor( + host: string, + port: number, + options: { serverInfo?: { agentName: string; url: string } } = {}, + ) { super(); this.host = host; this.port = port; + this.options = options; } override async start(): Promise { @@ -291,13 +304,26 @@ export class TcpSessionTransport extends SessionTransport { }; socket.once('error', onConnectError); }); + + if (this.options.serverInfo) { + const msg = new AgentDev.AgentDevMessage({ + message: { + case: 'serverInfo', + value: new AgentDev.ServerInfo(this.options.serverInfo), + }, + }); + await this.sendFramed(msg.toBinary()); + } } override async sendMessage(msg: pb.AgentSessionMessage): Promise { + await this.sendFramed(msg.toBinary()); + } + + private async sendFramed(data: Uint8Array): Promise { const socket = this.socket; if (this.closed || socket === null) return; - const data = msg.toBinary(); const header = Buffer.allocUnsafe(TCP_HEADER_SIZE); header.writeUInt32BE(data.length, 0); const flushed = socket.write(Buffer.concat([header, data]));