Amazon Bedrock AgentCore exploration

examples/aws-agentcore/README.md

@@ -0,0 +1,105 @@
+# Amazon Bedrock AgentCore Runtime Example
+
+This example demonstrates how to prepare a Pipecat bot for deployment to **Amazon Bedrock AgentCore Runtime** and enable it to invoke AgentCore tools.
+
+## Overview
+
+This example shows the set needed to:
+
+- Deploy your Pipecat bot to Amazon Bedrock AgentCore Runtime (which hosts and runs your bot)
+- Enable your bot to invoke AgentCore tools while running in the AgentCore Runtime
+
+The key additions to a standard Pipecat bot are the AgentCore-specific configurations and tool invocation handling that allow your bot to leverage the full AgentCore ecosystem.
+
+## Prerequisites
+
+- Accounts with:
+  - AWS
+  - OpenAI
+  - Deepgram
+  - Cartesia
+  - Daily
+- Python 3.10 or higher
+- `uv` package manager
+
+## IAM Configuration
+
+Configure your IAM user with the necessary policies for AgentCore usage. Start with these:
+
+- `BedrockAgentCoreFullAccess`
+- A new policy (maybe named `BedrockAgentCoreCLI`) configured [like this](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-permissions.html#runtime-permissions-starter-toolkit)
+
+You can also choose to specify more granular permissions; see [Amazon Bedrock AgentCore docs](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-permissions.html) for more information.
+
+To simplify the remaining steps in this README, it's a good idea to export some AWS-specific environment variables:
+
+```bash
+export AWS_SECRET_ACCESS_KEY=...
+export AWS_ACCESS_KEY_ID=...
+export AWS_REGION=...
+```
+
+## Agent Configuration
+
+Configure your bot as an AgentCore agent.
+
+```bash
+agentcore configure -e bot.py
+```
+
+Follow the prompts to complete the configuration.
+
+**IMPORTANT:** when asked if you want to use "Direct Code Deploy" or "Container", choose "Container". Today there is an incompatibility between Pipecat and "Direct Code Deploy".
+
+> For the curious: "Direct Code Deploy" requires that all bot dependencies have an `aarch64_manylinux2014` wheel...which is unfortunately not true for `numba`.
+
+## Deployment to AgentCore Runtime
+
+Deploy your configured bot to Amazon Bedrock AgentCore Runtime for production hosting.
+
+```bash
+agentcore launch --env OPENAI_API_KEY=... --env DEEPGRAM_API_KEY=... --env CARTESIA_API_KEY=... # -a <agent_name> (if multiple agents configured)
+```
+
+You should see commands related to tailing logs printed to the console. Copy and save them for later use.
+
+This is also the command you need to run after you've updated your bot code.
+
+## Running on AgentCore Runtime
+
+Run your bot on AgentCore Runtime.
+
+```bash
+agentcore invoke '{"roomUrl": "https://<your-domain>.daily.co/<room-name>"}' # -a <agent_name> (if multiple agents configured)
+```
+
+## Observation
+
+Paste the log tailing command you received when deploying your bot to AgentCore Runtime. It should look something like:
+
+```bash
+# Replace with your actual command
+aws logs tail /aws/bedrock-agentcore/runtimes/bot1-0uJkkT7QHC-DEFAULT --log-stream-name-prefix "2025/11/19/[runtime-logs]" --follow
+```
+
+## Running Locally
+
+You can also run your bot locally, using either the SmallWebRTC or Daily transport.
+
+First, copy `env.example` to `.env` and fill in the values.
+
+Then, run the bot:
+
+```bash
+# SmallWebRTC
+PIPECAT_LOCAL_DEV=1 uv run python bot.py
+
+# Daily
+PIPECAT_LOCAL_DEV=1 uv run python bot.py -t daily -d
+```
+
+> Ideally you should be able to use `agentcore launch --local`, but it doesn't currently appear to be working (even with [this workaround](https://github.com/aws/bedrock-agentcore-starter-toolkit/issues/156) applied), at least not for this project.
+
+## Additional Resources
+
+For a comprehensive guide to getting started with Amazon Bedrock AgentCore, including detailed setup instructions, see the [Amazon Bedrock AgentCore Developer Guide](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/what-is-bedrock-agentcore.html).

examples/aws-agentcore/bot.py

@@ -0,0 +1,184 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from bedrock_agentcore import BedrockAgentCoreApp
+from dotenv import load_dotenv
+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
+from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.runner.types import DailyRunnerArguments, RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+
+app = BedrockAgentCoreApp()
+
+load_dotenv(override=True)
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    await params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+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")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    # You can also register a function_name of None to get all functions
+    # sent to the same callback with an additional function_name parameter.
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
+
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    restaurant_function = FunctionSchema(
+        name="get_restaurant_recommendation",
+        description="Get a restaurant recommendation",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+        },
+        required=["location"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            context_aggregator.user(),
+            llm,
+            tts,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+@app.entrypoint
+async def bot(payload, context):
+    """Main bot entry point compatible with AWS Bedrock AgentCore Runtime."""
+    room_url = payload.get("roomUrl")
+    transport = await create_transport(
+        DailyRunnerArguments(room_url=room_url),
+        transport_params,
+    )
+    await run_bot(transport, RunnerArguments())
+
+
+if __name__ == "__main__":
+    # NOTE: ideally we shouldn't have to branch for local dev vs AgentCore, but
+    # local AgentCore container-based dev doesn't seem to be working, or at
+    # least not for this project.
+    if os.getenv("PIPECAT_LOCAL_DEV") == "1":
+        # Running locally
+        from pipecat.runner.run import main
+
+        main()
+    else:
+        # Running on AgentCore Runtime
+        app.run()

examples/aws-agentcore/env.example

@@ -0,0 +1,4 @@
+OPENAI_API_KEY=...
+DEEPGRAM_API_KEY=...
+CARTESIA_API_KEY=...
+DAILY_SAMPLE_ROOM_URL=https://<your-domain>.daily.co/<room-name>

examples/aws-agentcore/pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "agentcore-pipecat"
+version = "0.1.0"
+description = "Example for building Pipecat bots deployable to Amazon Bedrock AgentCore"
+requires-python = ">=3.10"
+dependencies = [
+    "bedrock-agentcore",
+    "pipecat-ai[webrtc,daily,silero,deepgram,openai,cartesia,local-smart-turn-v3,runner]",
+]
+
+[dependency-groups]
+dev = [
+    "bedrock-agentcore-starter-toolkit",
+    "pyright>=1.1.404,<2",
+    "ruff>=0.12.11,<1",
+]
+
+[tool.ruff]
+line-length = 100
+[tool.ruff.lint]
+select = ["I"]

examples/aws-agentcore/requirements.txt

@@ -0,0 +1,2 @@
+bedrock-agentcore
+pipecat-ai[webrtc,daily,silero,deepgram,openai,cartesia,local-smart-turn-v3,runner]