Agent Class

The Agent class is the central orchestrator that brings together all other components in the Vision Agents framework. It manages the conversation flow, handles real-time audio/video processing, coordinates responses, and integrates with external tools via MCP (Model Context Protocol) servers. It is the main interface for building AI-powered video and voice applications. It supports both traditional STT/TTS workflows and modern realtime speech-to-speech models, making it flexible for various use cases.

Do not reuse an Agent instance. Create a new agent for each call. Calling join() twice on the same agent raises RuntimeError.

from vision_agents.core import Agent, User
from vision_agents.plugins import openai, deepgram, elevenlabs, getstream

# Traditional STT/TTS mode
agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="AI Assistant", id="agent"),
    instructions="You're a helpful AI assistant",
    llm=openai.LLM(model="gpt-5.4"),
    stt=deepgram.STT(),
    tts=elevenlabs.TTS(),
)

# Realtime mode
agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="AI Assistant", id="agent"),
    instructions="You're a helpful AI assistant",
    llm=openai.Realtime(model="gpt-realtime", voice="marin"),
)

Constructor Parameters

• edge (EdgeTransport): The edge network provider for video & audio transport (you can choose any provider here)
• llm (LLM | AudioLLM | VideoLLM): The language model — can be a text LLM, audio-capable (AudioLLM), video-capable (VideoLLM), or a combined Realtime model
• agent_user (User): The agent’s user information (name, id, etc.)

Optional Parameters

• instructions (str): System instructions for the agent. Supports @file.md references to load instructions from markdown files (default: “Keep your replies short and dont use special characters.”)
• stt (Optional[STT]): Speech-to-text service (automatically disabled in realtime mode)
• tts (Optional[TTS]): Text-to-speech service (automatically disabled in realtime mode)
• turn_detection (Optional[TurnDetector]): Turn detection service for managing conversation turns. Automatically disabled in realtime mode. Ignored automatically if the configured STT plugin already provides built-in turn detection (when stt.turn_detection is True, e.g. elevenlabs.STT, deepgram.STT)
• processors (Optional[List[Processor]]): List of processors for video/audio processing
• avatar (Optional[Avatar]): Avatar provider for lip-synced video output (for example anam.Avatar(...))
• mcp_servers (Optional[List[MCPBaseServer]]): MCP servers for external tool access
• options (Optional[AgentOptions]): Configuration options — see AgentOptions below
• broadcast_metrics (bool): Broadcast agent metrics to call participants via custom events (default: False)
• broadcast_metrics_interval (float): Interval in seconds between metrics broadcasts (default: 5.0)
• multi_speaker_filter (Optional[AudioFilter]): Filter for multi-speaker audio routing. Defaults to FirstSpeakerWinsFilter, which locks onto the first active speaker and drops other participants’ audio until silence timeout or disconnect. Only activates when two or more participants have active audio tracks — see Multiple Speakers
• tracer (Tracer): OpenTelemetry tracer for distributed tracing (default: trace.get_tracer("agents"))
• profiler (Optional[Profiler]): Performance profiler for the agent and its plugins

AgentOptions

Field	Type	Default	Description
model_dir	str	System temp directory	Directory for downloaded model files (VAD, turn detection, etc.)

Mode Validation

When an AudioLLM is configured (currently Realtime models), STT, TTS, and turn detection are automatically disabled with a warning log. In non-realtime mode, the agent requires at least an audio processing path (STT, TTS, or turn detection) or video processors. Video-only agents without an LLM are allowed when video processors are configured.

Core Lifecycle Methods

async create_call(call_type: str, call_id: str) -> Call

Creates a call on the edge provider. Calls authenticate() automatically.

async join(call: Call, participant_wait_timeout: Optional[float] = 10.0) -> AsyncIterator[None]

Joins a video call. Must be called as an async context manager.

The agent can join the call only once. Once the call ends, the agent closes itself. join() always calls close() in its finally block.

Parameters

• call (Call): the call to join.
• participant_wait_timeout (Optional[float]): timeout in seconds to wait for other participants to join before proceeding.
  If 0, do not wait at all. If None, wait forever.
  Default - 10.0. Delegates to wait_for_participant().

call = await agent.create_call(call_type="default", call_id="my-call")
async with agent.join(call):
    await agent.simple_response("Say hi.")
    await agent.finish()  # Wait for call to end

async wait_for_participant(timeout: Optional[float] = None) -> None

Waits for at least one participant to join. Default None waits forever.

async finish()

Waits for the call to end gracefully. Subscribes to the call ended event. Returns immediately if not joined or already closed.

async close()

Cleans up all connections and resources. Safe to call multiple times. Called automatically when join() context exits.

async authenticate()

Authenticates the agent user with the edge provider. Idempotent — safe to call multiple times. Called automatically from join() and create_call(), so you usually don’t need to call it directly.

Response Methods

async simple_response(text: str, participant: Optional[Participant] = None, interrupt: bool = True)

Sends a text prompt to the active inference flow. The LLM generates a response which is then spoken through TTS (or realtime audio output). Use interrupt=True when you want this request to preempt an in-flight response.

await agent.simple_response("Hello, how can I help you?", interrupt=True)

async say(text: str, interrupt: bool = False)

Makes the agent speak text directly, bypassing the LLM. Works in transcribing (STT + LLM + TTS) mode only — the text is synthesized via TTS. In Realtime mode, say() logs a warning and does not produce audio (conversation history may still be updated). Use interrupt=True to stop current output before speaking.

await agent.say("Welcome to the call!", interrupt=False)

Monitoring Methods

idle_for() -> float

Returns the number of seconds the agent has been alone on the call (all human participants have left). Returns 0.0 while other participants are present or before join.

on_call_for() -> float

Returns the number of seconds since the agent joined the call.

Observability

metrics — AgentMetrics property for performance data. See Telemetry.

async send_metrics_event(event_type="agent_metrics", fields=None) — Broadcast metrics to call participants.

async send_custom_event(data: dict) — Push custom data to call participants via the edge provider.

MCP Integration

The Agent supports Model Context Protocol (MCP) for external tool integration. MCP servers connect when the agent joins a call:

from vision_agents.core import Agent, User
from vision_agents.core.mcp import MCPServerRemote

github_server = MCPServerRemote(
    url="https://api.githubcopilot.com/mcp/",
    headers={"Authorization": f"Bearer {github_pat}"}
)

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="Assistant", id="agent"),
    llm=...,
    mcp_servers=[github_server],
)

MCP tools are automatically registered with the LLM’s function registry and can be called during conversations. See the MCP guide for more.

Event System

The Agent makes it easy for developers to quickly subscribe and listen to events happening across all components. The event system merges all events across the plugin and core allowing you to listen to events in a single place using their respective type.

Core Events

• Agent Lifecycle Events: UserTurnStartedEvent, UserTurnEndedEvent, UserTranscriptEvent, AgentTurnStartedEvent, AgentTurnEndedEvent, AgentJoinedCallEvent, AgentLeftCallEvent, AgentFinishEvent (emitted only when finish() is cancelled, not on normal call end)
• Edge / Call Events: ParticipantJoinedEvent, ParticipantLeftEvent, CallEndedEvent, TrackAddedEvent, TrackRemovedEvent, AudioReceivedEvent
• LLM Events: LLMResponseFinalEvent, LLMErrorEvent
• Tool Events: ToolStartEvent, ToolEndEvent
• Realtime Events: RealtimeConnectedEvent, RealtimeDisconnectedEvent
• STT Events: STTConnectedEvent, STTDisconnectedEvent, STTErrorEvent
• TTS Events: TTSSynthesisStartEvent, TTSSynthesisCompleteEvent, TTSConnectedEvent, TTSDisconnectedEvent, TTSErrorEvent

See Events Reference for full field details.

Event Subscription

from vision_agents.core.edge.events import AudioReceivedEvent

@agent.events.subscribe
async def on_audio_received(event: AudioReceivedEvent):
    # Handle audio data
    pass

You can also use agent.subscribe(function) as an alternative to the decorator.

Debugging with local video files

For testing and debugging video processing without a live camera, you can use a local video file as the video source. This is useful for reproducible testing and development.

Video override only works when publish_video is True (requires an avatar or a video publisher processor). Set the path before calling join().

Using the CLI

Pass the --video-track-override option when running your agent:

uv run agent.py run --video-track-override=/path/to/video.mp4

Using the API

from vision_agents.core import Agent, User

agent = Agent(...)
agent.set_video_track_override_path("/path/to/video.mp4")

When a video override is set, the local video file plays in a loop at 30 FPS instead of any incoming video tracks from call participants. The track lifecycle remains intact (starts when a user joins, stops when they leave).