Agents

from vision_agents import Agent
from vision_agents.edge import getstream
from vision_agents.llm import gemini
from vision_agents.core.edge.types import User

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(id="agent-1", name="AI Assistant"),
    instructions="You are a helpful voice assistant. Keep responses concise.",
    llm=gemini.Realtime(),
    processors=[],  # Optional video/audio processors
)

# Join a call
async with agent.join(call):
    await agent.finish()  # Wait for call to end

from vision_agents.stt import deepgram
from vision_agents.tts import elevenlabs
from vision_agents.turn_detection import silero
from vision_agents.llm import openai

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(id="agent-1", name="AI Assistant"),
    instructions="You are a helpful assistant.",
    llm=openai.LLM(model="gpt-4o"),
    stt=deepgram.STT(),
    tts=elevenlabs.TTS(),
    turn_detection=silero.TurnDetector(),
)

async with agent.join(call):
    await agent.finish()

from vision_agents.edge import getstream

edge = getstream.Edge()

from my_processors import ObjectDetector, SentimentAnalyzer

agent = Agent(
    # ... other config
    processors=[
        ObjectDetector(),
        SentimentAnalyzer(),
    ]
)

from vision_agents.mcp import FileSystemServer, WebSearchServer

agent = Agent(
    # ... other config
    mcp_servers=[
        FileSystemServer(),
        WebSearchServer(),
    ]
)

async with agent.join(call, participant_wait_timeout=10.0) as session:
    # Agent is now connected
    await agent.simple_response("Hello! How can I help?")
    
    # Wait for call to end naturally
    await agent.finish()
# Agent automatically closes after context exits

async def _close(self):
    # Stop audio consumer
    # Stop metrics broadcast
    # Call stop() and close() on all plugins
    # Disconnect MCP servers
    # Stop video forwarders
    # Close RTC connection
    # Stop audio/video tracks

agent = Agent(
    # ... other config
    streaming_tts=True,  # Stream to TTS at sentence boundaries
)

from vision_agents.utils.audio_filter import FirstSpeakerWinsFilter

agent = Agent(
    # ... other config
    multi_speaker_filter=FirstSpeakerWinsFilter(),  # Default
)

agent = Agent(
    # ... other config
    broadcast_metrics=True,
    broadcast_metrics_interval=5.0,  # seconds
)

await agent.say(
    "I've processed your request.",
    user_id="agent-1",
    metadata={"action": "confirmation"}
)

from vision_agents.core.agents.agent_types import AgentOptions

options = AgentOptions(
    model_dir="/path/to/models",  # Directory for cached models
)

agent = Agent(
    # ... other config
    options=options
)

@agent.subscribe
async def on_llm_response(event: LLMResponseCompletedEvent):
    print(f"Agent said: {event.text}")

@agent.subscribe
async def on_user_transcript(event: STTTranscriptEvent):
    print(f"User said: {event.text}")

from opentelemetry import trace

tracer = trace.get_tracer("my-app")

agent = Agent(
    # ... other config
    tracer=tracer,
)

metrics = agent._metrics  # AgentMetrics instance
# Includes latency, token usage, error rates, etc.