Home - News ( United States | United Kingdom | Italy | Germany ) - Football scores

Showing content from https://docs.powertools.aws.dev/lambda/python/latest/core/event_handler/bedrock_agents/ below:

Amazon Bedrock Agents - Powertools for AWS Lambda (Python)

Create Amazon Bedrock Agents using event handlers with two different action groups approaches:

• OpenAPI schema
• Function details

flowchart LR
    Bedrock[LLM] <-- uses --> Agent
    You[User input] --> Agent
    Agent -- consults --> OpenAPI
    Agent -- consults --> Functions
    Agent[Amazon Bedrock Agents] -- invokes --> Lambda

    subgraph OpenAPI
        Schema
    end

    subgraph Functions
        ToolDescriptions[Tool Descriptions]
    end

    subgraph Lambda[Lambda Function]
        direction TB
        Parsing[Parameter Parsing] --> Validation
        Validation[Parameter Validation] --> Routing
        Routing --> Code[Your code]
        Code --> ResponseValidation[Response Validation]
        ResponseValidation --> ResponseBuilding[Response Building]
    end

    subgraph ActionGroup[Action Group]
        OpenAPI -. generated from .-> Lambda
        Functions -. defined in .-> Lambda
    end

    style Code fill:#ffa500,color:black,font-weight:bold,stroke-width:3px
    style You stroke:#0F0,stroke-width:2px

• Minimal boilerplate to build Amazon Bedrock Agents
• Support for both OpenAPI-based and Function-based actions
• Automatic generation of OpenAPI schemas from your business logic code
• Built-in data validation for requests and responses
• Similar experience to authoring REST and HTTP APIs

Data validation automatically validates the user input and the response of your AWS Lambda function against a set of constraints defined by you.

Event handler is a Powertools for AWS feature that processes an event, runs data parsing and validation, routes the request to a specific function, and returns a response to the caller in the proper format.

OpenAPI schema is an industry standard JSON-serialized string that represents the structure and parameters of your API.

Function details consist of a list of parameters, defined by their name, data type, and whether they are required. The agent uses these configurations to determine what information it needs to elicit from the user.

Action group is a collection of two resources where you define the actions that the agent should carry out when invoking your Lambda function.

Large Language Models (LLM) are very large deep learning models that are pre-trained on vast amounts of data, capable of extracting meanings from a sequence of text and understanding the relationship between words and phrases on it.

Amazon Bedrock Agent is an Amazon Bedrock feature to build and deploy conversational agents that can interact with your customers using Large Language Models (LLM) and AWS Lambda functions.

All examples shared in this documentation are available within the project repository

An action group defines actions that the agent can help the user perform. You can define action groups as OpenAPI-based or Function-based.

This is unnecessary if you're installing Powertools for AWS Lambda (Python) via Lambda Layer/SAR.

If you define the action group setting up an OpenAPI schema, you need to add pydantic as a dependency in your preferred tool e.g., requirements.txt, pyproject.toml. At this time, we only support Pydantic V2.

To create an agent, use the BedrockAgentResolver or the BedrockAgentFunctionResolver to annotate your actions. This is similar to the way all the other Event Handler resolvers work.

The resolvers used by Amazon Bedrock Agents are compatible with all Powertools for AWS Lambda features. For reference, we use Logger and Tracer in this example.

OpenAPI-based actions

Lambda handlerOpenAPI schemaInput payloadOutput payload

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from time import time

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentResolver()


@app.get("/current_time", description="Gets the current time in seconds")  # (1)!
@tracer.capture_method
def current_time() -> int:
    return int(time())


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)  # (2)!

1. description is a required field that should contain a human readable description of your action.
2. We take care of parsing, validating, routing and responding to the request.

Powertools for AWS Lambda generates this automatically from the Lambda handler.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

{
  "openapi": "3.0.3",
  "info": {
    "title": "Powertools API",
    "version": "1.0.0"
  },
  "servers": [
    {
      "url": "/"
    }
  ],
  "paths": {
    "/current_time": {
      "get": {
        "summary": "GET /current_time",
        "description": "Gets the current time in seconds",
        "operationId": "current_time_current_time_get",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "type": "integer",
                  "title": "Return"
                }
              }
            }
          },
          "422": {
            "description": "Validation Error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/HTTPValidationError"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "HTTPValidationError": {
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ValidationError"
            },
            "type": "array",
            "title": "Detail"
          }
        },
        "type": "object",
        "title": "HTTPValidationError"
      },
      "ValidationError": {
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "integer"
                }
              ]
            },
            "type": "array",
            "title": "Location"
          },
          "msg": {
            "type": "string",
            "title": "Message"
          },
          "type": {
            "type": "string",
            "title": "Error Type"
          }
        },
        "type": "object",
        "required": [
          "loc",
          "msg",
          "type"
        ],
        "title": "ValidationError"
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
  "sessionId": "123456789012345",
  "sessionAttributes": {},
  "inputText": "What is the current time?",
  "promptSessionAttributes": {},
  "apiPath": "/current_time",
  "agent": {
    "name": "TimeAgent",
    "version": "DRAFT",
    "id": "XLHH72XNF2",
    "alias": "TSTALIASID"
  },
  "httpMethod": "GET",
  "messageVersion": "1.0",
  "actionGroup": "CurrentTime"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "messageVersion": "1.0",
  "response": {
    "actionGroup": "CurrentTime",
    "apiPath": "/current_time",
    "httpMethod": "GET",
    "httpStatusCode": 200,
    "responseBody": {
      "application/json": {
        "body": "1704708165"
      }
    }
  }
}

Function-based actions

Lambda handlerInput payloadOutput payload

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from time import time

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentFunctionResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentFunctionResolver()


@app.tool(name="currentTime", description="Gets the current time in seconds")  # (1)!
@tracer.capture_method
def current_time() -> int:
    return int(time())


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)  # (2)!

1. name and description are optional here.
2. We take care of parsing, validating, routing and responding to the request.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
  "messageVersion": "1.0",
  "agent": {
    "name": "TimeAgent",
    "id": "XLHH72XNF2",
    "alias": "TSTALIASID",
    "version": "DRAFT"
  },
  "inputText": "What is the current time?",
  "sessionId": "123456789012345",
  "actionGroup": "CurrentTime",
  "function": "CurrentTime",
  "parameters": [],
  "sessionAttributes": {},
  "promptSessionAttributes": {}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
    "messageVersion": "1.0",
    "response": {
        "actionGroup": "CurrentTime",
        "function": "CurrentTime",
        "functionResponse": {
            "responseBody": {
                "application/json": {
                    "body": "1704708165"
                }
            }
        }
    }
}

The event sent by Amazon Bedrock Agents into your Lambda function contains a number of extra event fields, exposed in the app.current_event field.

You can for instance identify new conversations (session_id) or store and analyze entire conversations (input_text).

Accessing request fields

In this example, we append correlation data to all generated logs. This can be used to aggregate logs by session_id and observe the entire conversation between a user and the Agent.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from time import time

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
app = BedrockAgentResolver()


@app.get("/current_time", description="Gets the current time in seconds")
def current_time() -> int:
    logger.append_keys(
        session_id=app.current_event.session_id,
        action_group=app.current_event.action_group,
        input_text=app.current_event.input_text,
    )

    logger.info("Serving current_time")
    return int(time())


@logger.inject_lambda_context
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

The input event fields available depend on your Agent's configuration (OpenAPI-based or Function-based):

Use the get_openapi_json_schema function provided by the resolver to produce a JSON-serialized string that represents your OpenAPI schema. You can print this string or save it to a file. You'll use the file later when creating the Agent.

You'll need to regenerate the OpenAPI schema and update your Agent everytime your API changes.

app.pyOpenAPI schema

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from time import time

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentResolver()


@app.get("/current_time", description="Gets the current time in seconds")
@tracer.capture_method
def current_time() -> int:
    return int(time())


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)


if __name__ == "__main__":  # (1)!
    print(app.get_openapi_json_schema())  # (2)!

1. This ensures that it's only executed when running the file directly, and not when running on the Lambda runtime.
2. You can use additional options to customize the OpenAPI schema.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

{
  "openapi": "3.0.3",
  "info": {
    "title": "Powertools API",
    "version": "1.0.0"
  },
  "servers": [
    {
      "url": "/"
    }
  ],
  "paths": {
    "/current_time": {
      "get": {
        "summary": "GET /current_time",
        "description": "Gets the current time in seconds",
        "operationId": "current_time_current_time_get",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "type": "integer",
                  "title": "Return"
                }
              }
            }
          },
          "422": {
            "description": "Validation Error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/HTTPValidationError"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "HTTPValidationError": {
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ValidationError"
            },
            "type": "array",
            "title": "Detail"
          }
        },
        "type": "object",
        "title": "HTTPValidationError"
      },
      "ValidationError": {
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "integer"
                }
              ]
            },
            "type": "array",
            "title": "Location"
          },
          "msg": {
            "type": "string",
            "title": "Message"
          },
          "type": {
            "type": "string",
            "title": "Error Type"
          }
        },
        "type": "object",
        "required": [
          "loc",
          "msg",
          "type"
        ],
        "title": "ValidationError"
      }
    }
  }
}

To get the OpenAPI schema, run the Python script from your terminal. The script will generate the schema directly to standard output, which you can redirect to a file.

python3 app.py > schema.json

Working with Amazon Bedrock Agents will introduce non-deterministic behaviour to your system.

Amazon Bedrock uses LLMs to understand and respond to user input. These models are trained on vast amounts of data and are capable of extracting meanings from a sequence of text and understanding the relationship between words and phrases on it. However, this means that the same input can result in different outputs, depending on the characteristics of the LLM being used.

The OpenAPI schema provides context and semantics to the Agent that will support the decision process for invoking our Lambda function. Sparse or ambiguous schemas can result in unexpected outcomes.

We recommend enriching your OpenAPI schema with as many details as possible to help the Agent understand your functions, and make correct invocations. To achieve that, keep the following suggestions in mind:

• Always describe your function behaviour using the description field in your annotations
• When refactoring, update your description field to match the function outcomes
• Use distinct description for each function to have clear separation of semantics

You can define the expected format for incoming data and responses by using type annotations. Define constraints using standard Python types, dataclasses or Pydantic models. Pydantic is a popular library for data validation using Python type annotations.

The examples below uses Pydantic's EmailStr to validate the email address passed to the schedule_meeting function. The function then returns a boolean indicating if the meeting was successfully scheduled.

Lambda handlerOpenAPI schemaInput payloadOutput payload

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from pydantic import EmailStr
from typing_extensions import Annotated

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.event_handler.openapi.params import Body, Query
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentResolver()  # (1)!


@app.get("/schedule_meeting", description="Schedules a meeting with the team")
@tracer.capture_method
def schedule_meeting(
    email: Annotated[EmailStr, Query(description="The email address of the customer")],  # (2)!
) -> Annotated[bool, Body(description="Whether the meeting was scheduled successfully")]:  # (3)!
    logger.info("Scheduling a meeting", email=email)
    return True


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

1. No need to add the enable_validation parameter, as it's enabled by default.
2. Describe each input using human-readable descriptions
3. Add the typing annotations to your parameters and return types, and let the event handler take care of the rest

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

{
  "openapi": "3.0.3",
  "info": {
    "title": "Powertools API",
    "version": "1.0.0"
  },
  "servers": [
    {
      "url": "/"
    }
  ],
  "paths": {
    "/schedule_meeting": {
      "get": {
        "summary": "GET /schedule_meeting",
        "description": "Schedules a meeting with the team",
        "operationId": "schedule_meeting_schedule_meeting_get",
        "parameters": [
          {
            "description": "The email address of the customer",
            "required": true,
            "schema": {
              "type": "string",
              "format": "email",
              "title": "Email",
              "description": "The email address of the customer"
            },
            "name": "email",
            "in": "query"
          }
        ],
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "application/json": {
                "schema": {
                  "type": "boolean",
                  "title": "Return",
                  "description": "Whether the meeting was scheduled successfully"
                }
              }
            }
          },
          "422": {
            "description": "Validation Error",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/HTTPValidationError"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "HTTPValidationError": {
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ValidationError"
            },
            "type": "array",
            "title": "Detail"
          }
        },
        "type": "object",
        "title": "HTTPValidationError"
      },
      "ValidationError": {
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "string"
                },
                {
                  "type": "integer"
                }
              ]
            },
            "type": "array",
            "title": "Location"
          },
          "msg": {
            "type": "string",
            "title": "Message"
          },
          "type": {
            "type": "string",
            "title": "Error Type"
          }
        },
        "type": "object",
        "required": [
          "loc",
          "msg",
          "type"
        ],
        "title": "ValidationError"
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

{
  "sessionId": "123456789012345",
  "sessionAttributes": {},
  "inputText": "Schedule a meeting with the team. My email is foo@example.org",
  "promptSessionAttributes": {},
  "apiPath": "/schedule_meeting",
  "parameters": [
    {
      "name": "email",
      "type": "string",
      "value": "foo@example.org"
    }
  ],
  "agent": {
    "name": "TimeAgent",
    "version": "DRAFT",
    "id": "XLHH72XNF2",
    "alias": "TSTALIASID"
  },
  "httpMethod": "GET",
  "messageVersion": "1.0",
  "actionGroup": "SupportAssistant"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "messageVersion": "1.0",
  "response": {
    "actionGroup": "SupportAssistant",
    "apiPath": "/schedule_meeting",
    "httpMethod": "GET",
    "httpStatusCode": 200,
    "responseBody": {
      "application/json": {
        "body": "true"
      }
    }
  }
}

If the request validation fails, your event handler will not be called, and an error message is returned to Bedrock. Similarly, if the response fails validation, your handler will abort the response.

The event handler will always return a response according to the schema (OpenAPI) or type hints (Function-based). A validation failure in OpenAPI-based actions results in a 422 response. For both approaches, how Amazon Bedrock interprets that failure is non-deterministic, since it depends on the characteristics of the LLM being used.

OpenAPI-based Input payloadOpenAPI-based Output payload

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

{
  "sessionId": "123456789012345",
  "sessionAttributes": {},
  "inputText": "Schedule a meeting with the team. My email is foo@example@org",
  "promptSessionAttributes": {},
  "apiPath": "/schedule_meeting",
  "parameters": [
    {
      "name": "email",
      "type": "string",
      "value": "foo@example@org"
    }
  ],
  "agent": {
    "name": "TimeAgent",
    "version": "DRAFT",
    "id": "XLHH72XNF2",
    "alias": "TSTALIASID"
  },
  "httpMethod": "GET",
  "messageVersion": "1.0",
  "actionGroup": "SupportAssistant"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "messageVersion": "1.0",
  "response": {
    "actionGroup": "SupportAssistant",
    "apiPath": "/schedule_meeting",
    "httpMethod": "GET",
    "httpStatusCode": 200,
    "responseBody": {
      "application/json": {
        "body": "{\"statusCode\":422,\"detail\":[{\"loc\":[\"query\",\"email\"],\"type\":\"value_error.email\"}]}"
      }
    }
  }
}

sequenceDiagram
    Agent->>Lambda: input payload
    activate Lambda
    Lambda->>Parsing: parses input parameters
    Parsing->>Validation: validates input
    Validation-->Validation: failure
    box BedrockAgentResolver
    participant Lambda
    participant Parsing
    participant Validation
    participant Routing
    participant Your Code
    end
    Note right of Validation: Your code is never called
    Validation->>Agent: 422 response
    deactivate Lambda

To create an Agent for Amazon Bedrock, refer to the official documentation provided by AWS.

The following video demonstrates the end-to-end process:

During the creation process, you should use the schema previously generated when prompted for an OpenAPI specification.

To enrich the view that Amazon Bedrock Agents has of your Lambda functions, use a combination of Pydantic Models and OpenAPI type annotations to add constraints to your APIs parameters.

Adding constraints to your function parameters can help you to enforce data validation and improve the understanding of your APIs by Amazon Bedrock.

Whenever you use OpenAPI parameters to validate query strings or path parameters, you can enhance validation and OpenAPI documentation by using any of these parameters:

To implement these customizations, include extra constraints when defining your parameters:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import requests
from typing_extensions import Annotated

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.event_handler.openapi.params import Body, Query
from aws_lambda_powertools.utilities.typing import LambdaContext

app = BedrockAgentResolver()

logger = Logger()


@app.post(
    "/todos",
    description="Creates a TODO",
)
def create_todo(
    title: Annotated[str, Query(max_length=200, strict=True, description="The TODO title")],  # (1)!
) -> Annotated[bool, Body(description="Was the TODO created correctly?")]:
    todo = requests.post("https://jsonplaceholder.typicode.com/todos", data={"title": title})
    try:
        todo.raise_for_status()
        return True
    except Exception:
        logger.exception("Error creating TODO")
        return False


def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

1. Title should not be larger than 200 characters and strict mode is activated

Customize your API endpoints by adding metadata to endpoint definitions.

Here's a breakdown of various customizable fields:

To implement these customizations, include extra parameters when defining your routes:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import requests
from typing_extensions import Annotated

from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.event_handler.openapi.params import Body, Path
from aws_lambda_powertools.utilities.typing import LambdaContext

app = BedrockAgentResolver()


@app.get(
    "/todos/<todo_id>",
    summary="Retrieves a TODO item, returning it's title",
    description="Loads a TODO item identified by the `todo_id`",
    response_description="The TODO title",
    responses={
        200: {"description": "TODO item found"},
        404: {
            "description": "TODO not found",
        },
    },
    tags=["todos"],
)
def get_todo_title(
    todo_id: Annotated[int, Path(description="The ID of the TODO item from which to retrieve the title")],
) -> Annotated[str, Body(description="The TODO title")]:
    todo = requests.get(f"https://jsonplaceholder.typicode.com/todos/{todo_id}")
    todo.raise_for_status()

    return todo.json()["title"]


def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

You can enable user confirmation with Bedrock Agents to have your application ask for explicit user approval before invoking an action.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from time import time

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
app = BedrockAgentResolver()


@app.get(
    "/current_time",
    description="Gets the current time in seconds",
    openapi_extensions={"x-requireConfirmation": "ENABLED"},  # (1)!
)
def current_time() -> int:
    return int(time())


@logger.inject_lambda_context
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)


if __name__ == "__main__":
    print(app.get_openapi_json_schema())

1. Add an openapi extension

The default response only includes the essential fields to keep the payload size minimal, as AWS Lambda has a maximum response size of 25 KB.

You can use BedrockResponse class to add additional fields as needed, such as session attributes, prompt session attributes, and knowledge base configurations.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

from http import HTTPStatus

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.event_handler.api_gateway import BedrockResponse
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentResolver()


@app.get("/return_with_session", description="Returns a hello world with session attributes")
@tracer.capture_method
def hello_world():
    return BedrockResponse(
        status_code=HTTPStatus.OK.value,
        body={"message": "Hello from Bedrock!"},
        session_attributes={"user_id": "123"},
        prompt_session_attributes={"context": "testing"},
        knowledge_bases_configuration=[
            {
                "knowledgeBaseId": "kb-123",
                "retrievalConfiguration": {
                    "vectorSearchConfiguration": {"numberOfResults": 3, "overrideSearchType": "HYBRID"},
                },
            },
        ],
    )


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

Powertools handle the request from the Agent, parse, validate, and route it to the correct method in your code. The response is then validated and formatted back to the Agent.

sequenceDiagram
    actor User

    User->>Agent: What is the current time?
    Agent->>OpenAPI schema: consults
    OpenAPI schema-->>Agent: GET /current_time
    Agent-->>Agent: LLM interaction

    box Powertools
        participant Lambda
        participant Parsing
        participant Validation
        participant Routing
        participant Your Code
    end

    alt Function-based
        Agent->>Lambda: {function: "current_time", parameters: [], ...}
    end

    activate Lambda
    Lambda->>Parsing: parses parameters
    Parsing->>Validation: validates input
    Validation->>Routing: finds method to call
    Routing->>Your Code: executes
    activate Your Code
    Your Code->>Routing: 1709215709
    deactivate Your Code
    Routing->>Validation: returns output
    Validation->>Parsing: validates output
    Parsing->>Lambda: formats response

    alt Function-based
        Lambda->>Agent: {response: {functionResponse: {responseBody: {...}}}}
    end
    deactivate Lambda

    Agent-->>Agent: LLM interaction
    Agent->>User: "The current time is 14:08:29 GMT"

Test your routes by passing an Agent for Amazon Bedrock proxy event request:

assert_bedrock_agent_response.pyassert_bedrock_agent_response_module.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from dataclasses import dataclass

import assert_bedrock_agent_response_module
import pytest


@dataclass
class LambdaContext:
    function_name: str = "test"
    memory_limit_in_mb: int = 128
    invoked_function_arn: str = "arn:aws:lambda:eu-west-1:123456789012:function:test"
    aws_request_id: str = "da658bd3-2d6f-4e7b-8ec2-937234644fdc"


@pytest.fixture
def lambda_context() -> LambdaContext:
    return LambdaContext()


def test_lambda_handler(lambda_context: LambdaContext):
    minimal_event = {
        "apiPath": "/current_time",
        "httpMethod": "GET",
        "inputText": "What is the current time?",
    }
    # Example of Bedrock Agent API request event:
    # https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html#agents-lambda-input
    ret = assert_bedrock_agent_response_module.lambda_handler(minimal_event, lambda_context)
    assert ret["response"]["httpStatuScode"] == 200
    assert ret["response"]["responseBody"]["application/json"]["body"] != ""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import time

from typing_extensions import Annotated

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from aws_lambda_powertools.event_handler.openapi.params import Body
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = BedrockAgentResolver()


@app.get("/current_time", description="Gets the current time")
@tracer.capture_method
def current_time() -> Annotated[int, Body(description="Current time in milliseconds")]:
    return round(time.time() * 1000)


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

The BedrockAgentFunctionResolver streamlines agent function development through three core capabilities:

• Register Functions: Use the @app.tool() decorator to expose your functions to Bedrock Agents

• Process Parameters: Automatically maps input parameters from the agent to your function arguments
• Format Responses: Transforms your function outputs into properly structured Bedrock Agent responses

The default response only includes the essential fields to keep the payload size minimal, as AWS Lambda has a maximum response size of 25 KB.

You can use BedrockFunctionResponse class to customize your response with additional fields. This class allows you to:

• Return a response body
• Set session and prompt session attributes
• Set knowledge bases configurations
• Control the response state ("FAILURE" or "REPROMPT")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from aws_lambda_powertools.event_handler import BedrockAgentFunctionResolver, BedrockFunctionResponse
from aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext

app = BedrockAgentFunctionResolver()


@app.tool(description="Function that demonstrates response customization")
def custom_response():
    return BedrockFunctionResponse(
        body="Hello World",
        session_attributes={"user_id": "123"},
        prompt_session_attributes={"last_action": "greeting"},
        response_state="REPROMPT",
        knowledge_bases=[{"name": "kb1", "enabled": True}],
    )


def lambda_handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

RetroSearch is an open source project built by @garambo | Open a GitHub Issue

Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo

HTML: 3.2 | Encoding: UTF-8 | Version: 0.7.4