mypy

Author: anafalcao

aws_lambda_powertools/event_handler/bedrock_agent_function.py

@@ -29,12 +29,14 @@ def lambda_handler(event, context):
         return app.resolve(event, context)
     ```
     """
+
     def __init__(self) -> None:
         self._tools: dict[str, dict[str, Any]] = {}
         self.current_event: BedrockAgentFunctionEvent | None = None
 
     def tool(self, description: str | None = None) -> Callable:
         """Decorator to register a tool function"""
+
         def decorator(func: Callable) -> Callable:
             if not description:
                 raise ValueError("Tool description is required")
@@ -48,6 +50,7 @@ def decorator(func: Callable) -> Callable:
                 "description": description,
             }
             return func
+
         return decorator
 
     def resolve(self, event: dict[str, Any], context: Any) -> dict[str, Any]:
@@ -60,28 +63,27 @@ def resolve(self, event: dict[str, Any], context: Any) -> dict[str, Any]:
 
     def _resolve(self) -> dict[str, Any]:
         """Internal resolution logic"""
+        if self.current_event is None:
+            raise ValueError("No event to process")
+
         function_name = self.current_event.function
         action_group = self.current_event.action_group
 
         if function_name not in self._tools:
             return self._create_response(
                 action_group=action_group,
                 function_name=function_name,
-                result=f"Function not found: {function_name}"
+                result=f"Function not found: {function_name}",
             )
 
         try:
             result = self._tools[function_name]["function"]()
-            return self._create_response(
-                action_group=action_group,
-                function_name=function_name,
-                result=result
-            )
+            return self._create_response(action_group=action_group, function_name=function_name, result=result)
         except Exception as e:
             return self._create_response(
                 action_group=action_group,
                 function_name=function_name,
-                result=f"Error: {str(e)}"
+                result=f"Error: {str(e)}",
             )
 
     def _create_response(self, action_group: str, function_name: str, result: Any) -> dict[str, Any]:
@@ -91,12 +93,6 @@ def _create_response(self, action_group: str, function_name: str, result: Any) -
             "response": {
                 "actionGroup": action_group,
                 "function": function_name,
-                "functionResponse": {
-                    "responseBody": {
-                        "TEXT": {
-                            "body": str(result)
-                        }
-                    }
-                }
-            }
-        }
+                "functionResponse": {"responseBody": {"TEXT": {"body": str(result)}}},
+            },
+        }

tests/functional/event_handler/required_dependencies/test_bedrock_agent_functions.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import pytest
+
 from aws_lambda_powertools.event_handler import BedrockAgentFunctionResolver
 from aws_lambda_powertools.utilities.data_classes import BedrockAgentFunctionEvent
 from tests.functional.utils import load_event
@@ -74,6 +75,7 @@ def test_bedrock_agent_function_missing_description():
     # WHEN registering a tool without description
     # THEN raise ValueError
     with pytest.raises(ValueError, match="Tool description is required"):
+
         @app.tool()
         def test_function():
             return "test"
@@ -90,6 +92,7 @@ def test_function():
 
     # THEN raise ValueError on second registration
     with pytest.raises(ValueError, match="Tool 'test_function' already registered"):
+
         @app.tool(description="Second registration")
         def test_function():  # noqa: F811
             return "test"
@@ -106,4 +109,4 @@ def test_function():
     # WHEN calling with invalid event
     # THEN raise ValueError
     with pytest.raises(ValueError, match="Missing required field"):
-        app.resolve({}, {})
+        app.resolve({}, {})