Built-in HTTP Server

import logging
from dotenv import load_dotenv
from vision_agents.core import Agent, AgentLauncher, Runner, User
from vision_agents.plugins import deepgram, elevenlabs, gemini, getstream

load_dotenv()
logging.basicConfig(level=logging.INFO)


async def create_agent(**kwargs) -> Agent:
    """Factory function that creates and configures an agent."""
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Assistant", id="agent"),
        instructions="You are a helpful voice assistant.",
        llm=gemini.LLM("gemini-3.1-flash-lite-preview"),
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(eager_turn_detection=True),
    )

    @agent.llm.register_function(description="Get the current weather for a location")
    async def get_weather(location: str) -> str:
        return f"The weather in {location} is sunny and 72°F."

    return agent


async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    """Called when the agent should join a call."""
    call = await agent.create_call(call_type, call_id)

    async with agent.join(call):
        await agent.simple_response("Hello! How can I help you today?")
        await agent.finish()


if __name__ == "__main__":
    runner = Runner(AgentLauncher(create_agent=create_agent, join_call=join_call))
    runner.cli()

uv run  <your_agent.py> serve

uv run agent.py serve --host 0.0.0.0 --port 8000 --agents-log-level DEBUG

uv run <your_agent.py> run

curl -X POST http://127.0.0.1:8000/calls/my-call-123/sessions \
  -H "Content-Type: application/json" \
  -d '{"call_type": "default"}'

{
  "session_id": "abc-123",
  "call_id": "my-call-123",
  "session_started_at": "2025-01-15T10:30:00Z"
}

curl http://127.0.0.1:8000/calls/my-call-123/sessions/abc-123/metrics

{
  "session_id": "abc-123",
  "call_id": "my-call-123",
  "session_started_at": "2025-01-15T10:30:00Z",
  "metrics_generated_at": "2025-01-15T10:35:00Z",
  "metrics": {
    "llm_latency_ms__avg": 245.5,
    "llm_time_to_first_token_ms__avg": 120.3,
    "llm_input_tokens__total": 1500,
    "llm_output_tokens__total": 800,
    "stt_latency_ms__avg": 85.2,
    "tts_latency_ms__avg": 95.1
  }
}

from vision_agents.core import Runner, AgentLauncher, ServeOptions

runner = Runner(
    AgentLauncher(create_agent=create_agent, join_call=join_call),
    serve_options=ServeOptions(
        cors_allow_origins=["https://myapp.com"],
        cors_allow_methods=["GET", "POST", "DELETE"],
        cors_allow_headers=["Authorization"],
        cors_allow_credentials=True,
    ),
)

from fastapi import Header, HTTPException


async def can_start_session(
    call_id: str,
    authorization: str = Header(None),
) -> bool:
    """Check if the request is authorized to start a session."""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization required")
    user = await validate_token(authorization)
    if not user.has_permission("start_session"):
        raise HTTPException(status_code=403, detail="Permission denied")
    return True


async def can_close_session(
    call_id: str,
    authorization: str = Header(None),
) -> bool:
    """Check if the request is authorized to close a session."""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization required")
    user = await validate_token(authorization)
    if not user.can_access_call(call_id):
        raise HTTPException(status_code=403, detail="Cannot close this session")
    return True


runner = Runner(
    AgentLauncher(create_agent=create_agent, join_call=join_call),
    serve_options=ServeOptions(
        can_start_session=can_start_session,
        can_close_session=can_close_session,
    ),
)

from fastapi.middleware.gzip import GZipMiddleware

runner = Runner(AgentLauncher(create_agent=create_agent, join_call=join_call))

# Adding a custom endpoint
@runner.fast_api.get("/custom")
def custom_endpoint():
    return {"message": "Custom endpoint"}

# Add custom middleware
runner.fast_api.add_middleware(GZipMiddleware, minimum_size=1000)

from fastapi import FastAPI

app = FastAPI(
    title="My Agent Server",
    description="Custom agent server with additional features",
    version="1.0.0",
)

# Add your own routes before passing to Runner
@app.get("/custom")
def custom_endpoint():
    return {"message": "Custom endpoint"}

runner = Runner(
    AgentLauncher(create_agent=create_agent, join_call=join_call),
    serve_options=ServeOptions(fast_api=app),
)

runner = Runner(
    AgentLauncher(
        create_agent=create_agent,
        join_call=join_call,
        max_concurrent_sessions=10,       # Limit total concurrent agents
        max_sessions_per_call=1,          # One agent per call
        max_session_duration_seconds=3600, # 1 hour max per session
        agent_idle_timeout=120.0,         # Disconnect after 2 min alone
    )
)

from vision_agents.core import AgentLauncher

launcher = AgentLauncher(
    create_agent=create_agent,
    join_call=join_call,
    max_concurrent_sessions=10,
)

# Start the launcher (warmup + maintenance loop)
await launcher.start()

# Start a session — returns an AgentSession
session = await launcher.start_session(call_id="my-call-123", call_type="default")

# Query a session from the registry (works across nodes)
info = await launcher.get_session_info("my-call-123", session.id)

# Request session closure (works across nodes)
await launcher.request_close_session("my-call-123", session.id)

# Close a session running on this node
await launcher.close_session(session.id)

# Stop the launcher (closes all sessions + registry)
await launcher.stop()

import grpc
from vision_agents.core import AgentLauncher

class AgentService(agent_pb2_grpc.AgentServiceServicer):
    def __init__(self, launcher: AgentLauncher):
        self.launcher = launcher

    async def StartSession(self, request, context):
        session = await self.launcher.start_session(
            call_id=request.call_id,
            call_type=request.call_type,
        )
        return agent_pb2.StartSessionResponse(session_id=session.id)

    async def CloseSession(self, request, context):
        await self.launcher.request_close_session(
            call_id=request.call_id,
            session_id=request.session_id,
        )
        return agent_pb2.CloseSessionResponse()

# Start the launcher, then serve
launcher = AgentLauncher(create_agent=create_agent, join_call=join_call)
await launcher.start()

server = grpc.aio.server()
agent_pb2_grpc.add_AgentServiceServicer_to_server(AgentService(launcher), server)
server.add_insecure_port("[::]:50051")
await server.start()
await server.wait_for_termination()