wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(aws): use shared credential resolver in Polly, Bedrock, AgentCore

Browse Source 
Polly TTS, Bedrock LLM, and AgentCore previously did
`arg or os.getenv("AWS_...")` and handed the result straight to
aioboto3.  When only one of `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY`
was set, aioboto3 received a half-populated kwarg and errored instead of
falling through to the boto3 credential provider chain (instance
profiles, IRSA, ECS task roles, SSO, etc.).

Route credential resolution through the shared `resolve_credentials()`
helper introduced for AWS Transcribe so all four services follow the
same `explicit → env → boto3 chain` fallback.  Add an
`AWSCredentials.to_boto_kwargs()` method to bridge the dataclass field
names (`access_key`, `secret_key`) to the aioboto3 kwargs
(`aws_access_key_id`, `aws_secret_access_key`).

No public API changes.  Behaviour is identical for fully-explicit and
fully-env-var configurations; partial env vars now correctly trigger
the chain instead of erroring.
This commit is contained in:
Mark Backman
2026-05-04 19:23:53 -04:00
parent 35153de28e
commit 8becafee38
4 changed files with 53 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/pipecat/services/aws/agent_core.py
View File

@@ -12,7 +12,6 @@ Amazon Bedrock AgentCore Runtime and streams their responses as LLMTextFrames.
import asyncio import asyncio
import json import json
import os
from collections.abc import Callable from collections.abc import Callable
import aioboto3 import aioboto3
@@ -27,6 +26,7 @@ from pipecat.frames.frames import (
) )
from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.aws.utils import resolve_credentials
def default_context_to_payload_transformer( def default_context_to_payload_transformer(
@@ -122,8 +122,11 @@ class AWSAgentCoreProcessor(FrameProcessor):
Args: Args:
agentArn: The Amazon Web Services Resource Name (ARN) of the agent. agentArn: The Amazon Web Services Resource Name (ARN) of the agent.
aws_access_key: AWS access key ID. If None, uses default credentials. aws_access_key: AWS access key ID. If None, falls back to
aws_secret_key: AWS secret access key. If None, uses default credentials. environment variables and the default boto3 credential chain
(instance profiles, IRSA, ECS task roles, SSO, etc.).
aws_secret_key: AWS secret access key. Same fallback behaviour as
``aws_access_key``.
aws_session_token: AWS session token for temporary credentials. aws_session_token: AWS session token for temporary credentials.
aws_region: AWS region. aws_region: AWS region.
context_to_payload_transformer: Optional callable to transform context_to_payload_transformer: Optional callable to transform
@@ -139,13 +142,13 @@ class AWSAgentCoreProcessor(FrameProcessor):
self._agentArn = agentArn self._agentArn = agentArn
self._aws_session = aioboto3.Session() self._aws_session = aioboto3.Session()
# Store AWS session parameters for creating client in async context # Resolve credentials using the shared chain (explicit → env → boto3).
self._aws_params = { self._aws_params = resolve_credentials(
"aws_access_key_id": aws_access_key or os.getenv("AWS_ACCESS_KEY_ID"), aws_access_key_id=aws_access_key,
"aws_secret_access_key": aws_secret_key or os.getenv("AWS_SECRET_ACCESS_KEY"), aws_secret_access_key=aws_secret_key,
"aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"), aws_session_token=aws_session_token,
"region_name": aws_region or os.getenv("AWS_REGION", "us-east-1"), region=aws_region,
} ).to_boto_kwargs()
# Set transformers with defaults # Set transformers with defaults
self._context_to_payload_transformer = ( self._context_to_payload_transformer = (
@@ -204,7 +207,8 @@ class AWSAgentCoreProcessor(FrameProcessor):
# aioboto3's `client()` is an async context manager but its stubs don't # aioboto3's `client()` is an async context manager but its stubs don't
# advertise `__aenter__` / `__aexit__` in a way pyright can see. # advertise `__aenter__` / `__aexit__` in a way pyright can see.
async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues] async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues]
"bedrock-agentcore", **self._aws_params "bedrock-agentcore",
**self._aws_params, # pyright: ignore[reportArgumentType]
) as client: ) as client:
# Invoke the AgentCore agent # Invoke the AgentCore agent
response = await client.invoke_agent_runtime( response = await client.invoke_agent_runtime(

25
src/pipecat/services/aws/llm.py
View File

@@ -13,7 +13,6 @@ function calling.
import asyncio import asyncio
import json import json
import os
import re import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any from typing import Any
@@ -36,6 +35,7 @@ from pipecat.frames.frames import (
from pipecat.metrics.metrics import LLMTokenUsage from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.aws.utils import resolve_credentials
from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import LLMService
from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given
from pipecat.utils.tracing.service_decorators import traced_llm from pipecat.utils.tracing.service_decorators import traced_llm
@@ -135,8 +135,11 @@ class AWSBedrockLLMService(LLMService[AWSBedrockLLMAdapter]):
.. deprecated:: 0.0.105 .. deprecated:: 0.0.105
Use ``settings=AWSBedrockLLMService.Settings(model=...)`` instead. Use ``settings=AWSBedrockLLMService.Settings(model=...)`` instead.
aws_access_key: AWS access key ID. If None, uses default credentials. aws_access_key: AWS access key ID. If None, falls back to
aws_secret_key: AWS secret access key. If None, uses default credentials. environment variables and the default boto3 credential chain
(instance profiles, IRSA, ECS task roles, SSO, etc.).
aws_secret_key: AWS secret access key. Same fallback behaviour as
``aws_access_key``.
aws_session_token: AWS session token for temporary credentials. aws_session_token: AWS session token for temporary credentials.
aws_region: AWS region for the Bedrock service. aws_region: AWS region for the Bedrock service.
params: Model parameters and configuration. params: Model parameters and configuration.
@@ -215,14 +218,14 @@ class AWSBedrockLLMService(LLMService[AWSBedrockLLMAdapter]):
self._aws_session = aioboto3.Session() self._aws_session = aioboto3.Session()
# Store AWS session parameters for creating client in async context # Resolve credentials using the shared chain (explicit → env → boto3).
self._aws_params = { resolved = resolve_credentials(
"aws_access_key_id": aws_access_key or os.getenv("AWS_ACCESS_KEY_ID"), aws_access_key_id=aws_access_key,
"aws_secret_access_key": aws_secret_key or os.getenv("AWS_SECRET_ACCESS_KEY"), aws_secret_access_key=aws_secret_key,
"aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"), aws_session_token=aws_session_token,
"region_name": aws_region or os.getenv("AWS_REGION", "us-east-1"), region=aws_region,
"config": client_config, )
} self._aws_params = {**resolved.to_boto_kwargs(), "config": client_config}
self._retry_timeout_secs = retry_timeout_secs self._retry_timeout_secs = retry_timeout_secs
self._retry_on_timeout = retry_on_timeout self._retry_on_timeout = retry_on_timeout

26
src/pipecat/services/aws/tts.py
View File

@@ -10,7 +10,6 @@ This module provides integration with Amazon Polly for text-to-speech synthesis,
supporting multiple languages, voices, and SSML features. supporting multiple languages, voices, and SSML features.
""" """
import os
from collections.abc import AsyncGenerator from collections.abc import AsyncGenerator
from dataclasses import dataclass, field from dataclasses import dataclass, field
@@ -23,6 +22,7 @@ from pipecat.frames.frames import (
Frame, Frame,
TTSAudioRawFrame, TTSAudioRawFrame,
) )
from pipecat.services.aws.utils import resolve_credentials
from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven
from pipecat.services.tts_service import TTSService from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language, resolve_language from pipecat.transcriptions.language import Language, resolve_language
@@ -191,8 +191,11 @@ class AWSPollyTTSService(TTSService):
"""Initializes the AWS Polly TTS service. """Initializes the AWS Polly TTS service.
Args: Args:
api_key: AWS secret access key. If None, uses AWS_SECRET_ACCESS_KEY environment variable. api_key: AWS secret access key. If None, falls back to environment
aws_access_key_id: AWS access key ID. If None, uses AWS_ACCESS_KEY_ID environment variable. variables and the default boto3 credential chain (instance
profiles, IRSA, ECS task roles, SSO, etc.).
aws_access_key_id: AWS access key ID. Same fallback behaviour as
``api_key``.
aws_session_token: AWS session token for temporary credentials. aws_session_token: AWS session token for temporary credentials.
region: AWS region for Polly service. Defaults to 'us-east-1'. region: AWS region for Polly service. Defaults to 'us-east-1'.
voice_id: Voice ID to use for synthesis. Defaults to 'Joanna'. voice_id: Voice ID to use for synthesis. Defaults to 'Joanna'.
@@ -250,13 +253,13 @@ class AWSPollyTTSService(TTSService):
**kwargs, **kwargs,
) )
# Get credentials from environment variables if not provided # Resolve credentials using the shared chain (explicit → env → boto3).
self._aws_params = { self._aws_params = resolve_credentials(
"aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"), aws_access_key_id=aws_access_key_id,
"aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"), aws_secret_access_key=api_key,
"aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"), aws_session_token=aws_session_token,
"region_name": region or os.getenv("AWS_REGION", "us-east-1"), region=region,
} ).to_boto_kwargs()
self._aws_session = aioboto3.Session() self._aws_session = aioboto3.Session()
@@ -348,7 +351,8 @@ class AWSPollyTTSService(TTSService):
# aioboto3's `client()` is an async context manager but its stubs # aioboto3's `client()` is an async context manager but its stubs
# don't advertise `__aenter__` / `__aexit__` to pyright. # don't advertise `__aenter__` / `__aexit__` to pyright.
async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues] async with self._aws_session.client( # pyright: ignore[reportGeneralTypeIssues]
"polly", **self._aws_params "polly",
**self._aws_params, # pyright: ignore[reportArgumentType]
) as polly: ) as polly:
response = await polly.synthesize_speech(**filtered_params) response = await polly.synthesize_speech(**filtered_params)
if "AudioStream" in response: if "AudioStream" in response:

9
src/pipecat/services/aws/utils.py
View File

@@ -34,6 +34,15 @@ class AWSCredentials:
session_token: str | None session_token: str | None
region: str region: str
def to_boto_kwargs(self) -> dict[str, str | None]:
"""Return credentials as kwargs accepted by ``boto3``/``aioboto3`` clients."""
return {
"aws_access_key_id": self.access_key,
"aws_secret_access_key": self.secret_key,
"aws_session_token": self.session_token,
"region_name": self.region,
}
def resolve_credentials( def resolve_credentials(
*, *,