Provide a way to run aws-agentcore-websocket example locally before deploying it to AgentCore Runtime

Author: kompfner

deployment/aws-agentcore-websocket/README.md

@@ -62,7 +62,7 @@ You can also choose to specify more granular permissions; see [Amazon Bedrock Ag
    cp env.example .env
    ```
 
-   Add your AWS credentials and configuration, for generating the signed WebSocket URL in the `/start` endpoint:
+   Add your AWS credentials and configuration, for generating a signed WebSocket URL in the `/start` endpoint:
 
    - `AWS_ACCESS_KEY_ID`
    - `AWS_SECRET_ACCESS_KEY`
@@ -113,7 +113,7 @@ This is also the command you need to run after you've updated your agent code.
 
 ## Running the Server
 
-The server provides a `/start` endpoint that generates signed WebSocket URLs for the client.
+The server provides a `/start` endpoint that generates WebSocket URLs for the client to connect to the agent.
 
 See [the server README](./server/README.md) for setup and run instructions.
 
@@ -123,6 +123,27 @@ Once the server is running, you can run the client to connect to your AgentCore-
 
 See [the client README](./client/README.md) for setup and run instructions.
 
+## Testing Locally
+
+To test agent logic locally before deploying to AgentCore Runtime, do the following.
+
+First, run the agent locally:
+
+```bash
+cd agent
+uv run python agent.py
+```
+
+This will make the agent reachable at "ws://localhost:8080/ws".
+
+Then, run the server as usual, but with the `LOCAL_AGENT=1` environment variable:
+
+```bash
+LOCAL_AGENT=1 uv run python server.py
+```
+
+You can then [run your client as usual](#running-the-client).
+
 ## Observation
 
 Paste one of the log tailing commands you received when deploying your agent to AgentCore Runtime. It should look something like:

deployment/aws-agentcore-websocket/agent/agent.py

@@ -11,7 +11,6 @@
 from loguru import logger
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
@@ -27,7 +26,7 @@
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
 from pipecat.services.llm_service import FunctionCallParams
-from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.base_transport import BaseTransport
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams, FastAPIWebsocketTransport
 
@@ -44,25 +43,6 @@ async def fetch_restaurant_recommendation(params: FunctionCallParams):
     await params.result_callback({"name": "The Golden Dragon"})
 
 
-# We store functions so objects (e.g. SileroVADAnalyzer) don't get
-# instantiated. The function will be called when the desired transport gets
-# selected.
-transport_params = {
-    "daily": lambda: DailyParams(
-        audio_in_enabled=True,
-        audio_out_enabled=True,
-        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
-        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
-    ),
-    "webrtc": lambda: TransportParams(
-        audio_in_enabled=True,
-        audio_out_enabled=True,
-        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
-        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
-    ),
-}
-
-
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
@@ -176,7 +156,8 @@ async def agentcore_bot(websocket, context):
             audio_in_enabled=True,
             audio_out_enabled=True,
             add_wav_header=False,
-            vad_analyzer=SileroVADAnalyzer(),
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+            turn_analyzer=LocalSmartTurnAnalyzerV3(),
             serializer=ProtobufFrameSerializer(),
         ),
     )

deployment/aws-agentcore-websocket/server/README.md

@@ -1,6 +1,6 @@
 # Server
 
-This server provides a `/start` endpoint that generates signed WebSocket URLs for connecting to the agent running on Amazon Bedrock AgentCore.
+This server provides a `/start` endpoint that generates WebSocket URLs for connecting to the agent running on Amazon Bedrock AgentCore.
 
 ## Prerequisites
 
@@ -35,11 +35,19 @@ uv run python server.py
 
 The server will be available at `http://localhost:7860`.
 
+## Running the Server in Local Agent Mode
+
+If you want to test a locally-running agent (reachable at "ws://localhost:8080/ws"), start the server like this:
+
+```bash
+LOCAL_AGENT=1 uv run python server.py
+```
+
 ## Endpoint
 
 ### POST /start
 
-Returns a signed WebSocket URL for the client to connect to the agent running on Amazon Bedrock AgentCore.
+Returns a WebSocket URL for the client to connect to the agent running on Amazon Bedrock AgentCore.
 
 **Response:**
 

deployment/aws-agentcore-websocket/server/server.py

@@ -47,6 +47,10 @@ async def start_bot(request: Request) -> Dict[Any, Any]:
     Returns:
         Dict[Any, Any]: Contains the signed WebSocket URL for the client to connect
     """
+    # Check if LOCAL_AGENT mode is enabled
+    if os.getenv("LOCAL_AGENT") == "1":
+        return {"ws_url": "ws://localhost:8080/ws"}
+
     # Get required environment variables
     access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
     secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")