diff --git a/examples/aws-agentcore/server/server.py b/examples/aws-agentcore/server/server.py new file mode 100644 index 000000000..d798a353c --- /dev/null +++ b/examples/aws-agentcore/server/server.py @@ -0,0 +1,152 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import sys +import uuid +from contextlib import asynccontextmanager +from http import HTTPMethod +from typing import Any, Dict, List, Optional, TypedDict, Union + +import uvicorn +from fastapi import BackgroundTasks, FastAPI, Request, Response +from fastapi.responses import RedirectResponse +from loguru import logger +from pipecat.transports.smallwebrtc.request_handler import ( + IceCandidate, + SmallWebRTCPatchRequest, + SmallWebRTCRequest, +) +from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + +app = FastAPI() + +# Mount the frontend at / +app.mount("/client", SmallWebRTCPrebuiltUI) + +# In-memory store of active sessions: session_id -> session info +active_sessions: Dict[str, Dict[str, Any]] = {} + + +@app.get("/", include_in_schema=False) +async def root_redirect(): + return RedirectResponse(url="/client/") + + +@app.post("/api/offer") +async def offer(request: SmallWebRTCRequest): + """Handle WebRTC offer requests via SmallWebRTCRequestHandler.""" + print(f"Received offer:{request}") + # TODO need to send this to agentcore + answer = None + return answer + + +@app.patch("/api/offer") +async def ice_candidate(request: SmallWebRTCPatchRequest): + """Handle WebRTC new ice candidate requests.""" + logger.debug(f"Received patch request: {request}") + # TODO need to send this to agentcore + return {"status": "success"} + + +@app.post("/start") +async def rtvi_start(request: Request): + """Mimic Pipecat Cloud's /start endpoint.""" + + class IceServer(TypedDict, total=False): + urls: Union[str, List[str]] + + class IceConfig(TypedDict): + iceServers: List[IceServer] + + class StartBotResult(TypedDict, total=False): + sessionId: str + iceConfig: Optional[IceConfig] + + # Parse the request body + try: + request_data = await request.json() + logger.debug(f"Received request: {request_data}") + except Exception as e: + logger.error(f"Failed to parse request body: {e}") + request_data = {} + + # Store session info immediately in memory, replicate the behavior expected on Pipecat Cloud + session_id = str(uuid.uuid4()) + active_sessions[session_id] = request_data + + result: StartBotResult = {"sessionId": session_id} + if request_data.get("enableDefaultIceServers"): + result["iceConfig"] = IceConfig( + iceServers=[IceServer(urls=["stun:stun.l.google.com:19302"])] + ) + + return result + + +@app.api_route( + "/sessions/{session_id}/{path:path}", + methods=["GET", "POST", "PUT", "PATCH", "DELETE"], +) +async def proxy_request( + session_id: str, path: str, request: Request, background_tasks: BackgroundTasks +): + """Mimic Pipecat Cloud's proxy.""" + active_session = active_sessions.get(session_id) + if active_session is None: + return Response(content="Invalid or not-yet-ready session_id", status_code=404) + + if path.endswith("api/offer"): + # Parse the request body and convert to SmallWebRTCRequest + try: + request_data = await request.json() + if request.method == HTTPMethod.POST.value: + webrtc_request = SmallWebRTCRequest( + sdp=request_data["sdp"], + type=request_data["type"], + pc_id=request_data.get("pc_id"), + restart_pc=request_data.get("restart_pc"), + request_data=request_data, + ) + return await offer(webrtc_request, background_tasks) + elif request.method == HTTPMethod.PATCH.value: + patch_request = SmallWebRTCPatchRequest( + pc_id=request_data["pc_id"], + candidates=[IceCandidate(**c) for c in request_data.get("candidates", [])], + ) + return await ice_candidate(patch_request) + except Exception as e: + logger.error(f"Failed to parse WebRTC request: {e}") + return Response(content="Invalid WebRTC request", status_code=400) + + logger.info(f"Received request for path: {path}") + return Response(status_code=200) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + yield # Run app + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="WebRTC demo") + parser.add_argument( + "--host", default="localhost", help="Host for HTTP server (default: localhost)" + ) + parser.add_argument( + "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)" + ) + parser.add_argument("--verbose", "-v", action="count") + args = parser.parse_args() + + logger.remove(0) + if args.verbose: + logger.add(sys.stderr, level="TRACE") + else: + logger.add(sys.stderr, level="DEBUG") + + uvicorn.run(app, host=args.host, port=args.port)