Edge Networks

from vision_agents.core.edge import EdgeTransport
from vision_agents.core.edge.types import User, Connection
from vision_agents.core.edge.call import Call

class CustomEdge(EdgeTransport):
    async def authenticate(self, user: User) -> None:
        """Authenticate agent user with the transport."""
        
    async def create_call(self, call_id: str, **kwargs) -> Call:
        """Create or retrieve a call."""
        
    async def join(self, agent: Agent, call: Call, **kwargs) -> Connection:
        """Join a call and establish connection."""
        
    async def publish_tracks(
        self,
        audio_track: Optional[MediaStreamTrack],
        video_track: Optional[MediaStreamTrack],
    ):
        """Publish audio/video tracks to the call."""
        
    async def close(self):
        """Clean up all resources."""

from vision_agents.core.edge.events import (
    AudioReceivedEvent,
    TrackAddedEvent,
    TrackRemovedEvent,
    CallEndedEvent,
)

class EdgeTransport:
    def __init__(self):
        self.events = EventManager()
        self.events.register(
            AudioReceivedEvent,     # Audio received from participant
            TrackAddedEvent,        # Media track added to call
            TrackRemovedEvent,      # Media track removed
            CallEndedEvent,         # Call ended
        )

from vision_agents.core.edge.types import User

agent_user = User(
    id="agent-123",
    name="AI Assistant",
    image="https://example.com/avatar.png",  # Optional
)

from vision_agents.core.edge.types import Participant

@dataclass
class Participant:
    original: Any            # Provider-specific participant object
    user_id: str            # User identifier (not necessarily unique)
    id: str                 # Unique participant ID for this call

class Connection(abc.ABC):
    async def close(self) -> None:
        """Close the connection and clean up."""
    
    async def wait_for_participant(
        self, 
        timeout: Optional[float] = None
    ) -> None:
        """Wait for a participant to join."""
    
    def idle_since(self) -> float:
        """Timestamp when all participants left (or 0.0 if active)."""

class TrackType(enum.IntEnum):
    UNSPECIFIED = 0
    AUDIO = 1
    VIDEO = 2
    SCREEN_SHARE = 3
    SCREEN_SHARE_AUDIO = 4

from vision_agents.edge import getstream
from vision_agents.core.edge.types import User

# Initialize edge
edge = getstream.Edge()

# Create agent user
agent_user = User(
    id="agent-1",
    name="Voice Assistant",
)

# Create agent with edge
from vision_agents import Agent

agent = Agent(
    edge=edge,
    agent_user=agent_user,
    # ... other config
)

# Agent handles authentication automatically
call = await agent.create_call(
    call_type="audio_room",
    call_id="support-session-123",
)

async with agent.join(call, participant_wait_timeout=10.0):
    # Agent is now connected to the call
    await agent.finish()

async with agent.join(call, participant_wait_timeout=30.0):
    # Waits up to 30 seconds for a participant
    # Proceeds after timeout even if no one joins
    await agent.finish()

# Or wait manually
await agent.wait_for_participant(timeout=60.0)

idle_duration = agent.idle_for()

if idle_duration > 300:  # 5 minutes
    print("No participants for 5 minutes, closing call")
    await agent.close()

duration = agent.on_call_for()
print(f"Agent has been on call for {duration:.1f} seconds")

# In agent.join():
audio_track = self._audio_track if self.publish_audio else None
video_track = self._video_track if self.publish_video else None

if audio_track or video_track:
    await self.edge.publish_tracks(audio_track, video_track)

@edge.events.subscribe
async def on_video_track_added(event: TrackAddedEvent):
    if event.track_type in (TrackType.VIDEO, TrackType.SCREEN_SHARE):
        # Subscribe to video track
        track = edge.add_track_subscriber(event.track_id)
        # Process with video processors or LLM

# From TrackInfo
priority = 1 if track_type == TrackType.SCREEN_SHARE else 0

@edge.events.subscribe
async def on_audio_received(event: AudioReceivedEvent):
    if event.pcm_data and event.participant:
        # Audio is buffered per-participant
        # Agent processes audio in 20ms chunks
        await process_audio_chunk(event.pcm_data)

# Internal implementation
self._participant_queues: dict[str, tuple[Participant, AudioQueue]] = {}

await edge.send_custom_event({
    "type": "agent_metric",
    "latency_ms": 120,
    "confidence": 0.95,
})

# Agent handles this automatically during join:
self.conversation = await self.edge.create_conversation(
    call,
    self.agent_user,
    self.instructions.full_reference,
)

# Provide conversation to LLM
self.llm.set_conversation(self.conversation)

edge.open_demo(
    call_id="my-call-123",
    token="user-token",
)

try:
    async with agent.join(call):
        await agent.finish()
except ConnectionError as e:
    logger.error(f"Failed to connect: {e}")
    # Retry logic or fallback
except asyncio.TimeoutError:
    logger.error("Connection timeout")

@edge.events.subscribe
async def on_call_ended(event: CallEndedEvent):
    logger.info("Call ended, cleaning up")
    await agent.close()

from vision_agents.core.edge import EdgeTransport
from vision_agents.core.edge.types import User, Connection
from vision_agents.core.edge.call import Call

class MyCustomEdge(EdgeTransport):
    async def authenticate(self, user: User) -> None:
        # Authenticate with your service
        self.token = await my_service.authenticate(user)
    
    async def create_call(self, call_id: str, **kwargs) -> Call:
        # Create call via your API
        response = await my_service.create_call(call_id)
        return Call(id=call_id, response=response)
    
    async def join(self, agent, call, **kwargs) -> Connection:
        # Establish WebRTC connection
        connection = await my_service.join(call.id, self.token)
        
        # Emit events as audio arrives
        self.events.send(AudioReceivedEvent(...))
        
        return MyConnection(connection)
    
    async def publish_tracks(self, audio_track, video_track):
        # Publish tracks to the call
        await my_service.publish(audio_track, video_track)
    
    async def close(self):
        # Clean up resources
        await my_service.disconnect()