> ## Documentation Index
> Fetch the complete documentation index at: https://visionagents.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

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

```python theme={null}
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](#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](/guides/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()`.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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](/core/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:

```python theme={null}
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](/guides/mcp-tool-calling) 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](/reference/events-reference) for full field details.

### Event Subscription

```python theme={null}
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.

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

### Using the CLI

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

```bash theme={null}
uv run agent.py run --video-track-override=/path/to/video.mp4
```

### Using the API

```python theme={null}
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).