diff --git a/aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py b/aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py index 399c435b3ec..45f3cd81f1f 100644 --- a/aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py +++ b/aws_lambda_powertools/utilities/data_classes/bedrock_agent_event.py @@ -1,3 +1,4 @@ +from functools import cached_property from typing import Any, Dict, List, Optional from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper @@ -112,3 +113,16 @@ def query_string_parameters(self) -> Optional[Dict[str, str]]: @property def resolved_headers_field(self) -> Optional[Dict[str, Any]]: return {} + + @cached_property + def json_body(self) -> Any: + # In Bedrock Agent events, body parameters are encoded differently + # @see https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html#agents-lambda-input + if not self.request_body: + return None + + json_body = self.request_body.content.get("application/json") + if not json_body: + return None + + return {x.name: x.value for x in json_body.properties} diff --git a/tests/functional/event_handler/test_bedrock_agent.py b/tests/functional/event_handler/test_bedrock_agent.py index 9f9a64427bf..74e91759dc0 100644 --- a/tests/functional/event_handler/test_bedrock_agent.py +++ b/tests/functional/event_handler/test_bedrock_agent.py @@ -2,7 +2,9 @@ from typing import Any, Dict from aws_lambda_powertools.event_handler import BedrockAgentResolver, Response, content_types +from aws_lambda_powertools.event_handler.openapi.params import Body from aws_lambda_powertools.event_handler.openapi.pydantic_loader import PYDANTIC_V2 +from aws_lambda_powertools.shared.types import Annotated from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent from tests.functional.utils import load_event @@ -157,3 +159,28 @@ def claims(): body = result["response"]["responseBody"]["text/plain"]["body"] assert body == "Something went wrong" + + +def test_bedrock_agent_with_post(): + # GIVEN a Bedrock Agent resolver with a POST method + app = BedrockAgentResolver() + + @app.post("/send-reminders", description="Sends reminders") + def send_reminders( + _claim_id: Annotated[int, Body(description="Claim ID", alias="claimId")], + _pending_documents: Annotated[str, Body(description="Social number and VAT", alias="pendingDocuments")], + ) -> Annotated[bool, Body(description="returns true if I like the email")]: + return True + + # WHEN calling the event handler + result = app(load_event("bedrockAgentPostEvent.json"), {}) + + # THEN process the event correctly + assert result["messageVersion"] == "1.0" + assert result["response"]["apiPath"] == "/send-reminders" + assert result["response"]["httpMethod"] == "POST" + assert result["response"]["httpStatusCode"] == 200 + + # THEN return the correct result + body = result["response"]["responseBody"]["application/json"]["body"] + assert json.loads(body) is True