Module aws_lambda_powertools.utilities.data_classes.bedrock_agent_event

Expand source code
from typing import Dict, List, Optional

from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper


class BedrockAgentInfo(DictWrapper):
    @property
    def name(self) -> str:
        return self["name"]

    @property
    def id(self) -> str:  # noqa: A003
        return self["id"]

    @property
    def alias(self) -> str:
        return self["alias"]

    @property
    def version(self) -> str:
        return self["version"]


class BedrockAgentProperty(DictWrapper):
    @property
    def name(self) -> str:
        return self["name"]

    @property
    def type(self) -> str:  # noqa: A003
        return self["type"]

    @property
    def value(self) -> str:
        return self["value"]


class BedrockAgentRequestMedia(DictWrapper):
    @property
    def properties(self) -> List[BedrockAgentProperty]:
        return [BedrockAgentProperty(x) for x in self["properties"]]


class BedrockAgentRequestBody(DictWrapper):
    @property
    def content(self) -> Dict[str, BedrockAgentRequestMedia]:
        return {k: BedrockAgentRequestMedia(v) for k, v in self["content"].items()}


class BedrockAgentEvent(BaseProxyEvent):
    """
    Bedrock Agent input event

    See https://docs.aws.amazon.com/bedrock/latest/userguide/agents-create.html
    """

    @property
    def message_version(self) -> str:
        return self["messageVersion"]

    @property
    def input_text(self) -> str:
        return self["inputText"]

    @property
    def session_id(self) -> str:
        return self["sessionId"]

    @property
    def action_group(self) -> str:
        return self["actionGroup"]

    @property
    def api_path(self) -> str:
        return self["apiPath"]

    @property
    def http_method(self) -> str:
        return self["httpMethod"]

    @property
    def parameters(self) -> Optional[List[BedrockAgentProperty]]:
        return [BedrockAgentProperty(x) for x in self["parameters"]] if self.get("parameters") else None

    @property
    def request_body(self) -> Optional[BedrockAgentRequestBody]:
        return BedrockAgentRequestBody(self["requestBody"]) if self.get("requestBody") else None

    @property
    def agent(self) -> BedrockAgentInfo:
        return BedrockAgentInfo(self["agent"])

    @property
    def session_attributes(self) -> Dict[str, str]:
        return self["sessionAttributes"]

    @property
    def prompt_session_attributes(self) -> Dict[str, str]:
        return self["promptSessionAttributes"]

    # The following methods add compatibility with BaseProxyEvent
    @property
    def path(self) -> str:
        return self["apiPath"]

    @property
    def query_string_parameters(self) -> Optional[Dict[str, str]]:
        # In Bedrock Agent events, query string parameters are passed as undifferentiated parameters,
        # together with the other parameters. So we just return all parameters here.
        return {x["name"]: x["value"] for x in self["parameters"]} if self.get("parameters") else None

Classes

class BedrockAgentEvent (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Bedrock Agent input event

See https://docs.aws.amazon.com/bedrock/latest/userguide/agents-create.html

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code
class BedrockAgentEvent(BaseProxyEvent):
    """
    Bedrock Agent input event

    See https://docs.aws.amazon.com/bedrock/latest/userguide/agents-create.html
    """

    @property
    def message_version(self) -> str:
        return self["messageVersion"]

    @property
    def input_text(self) -> str:
        return self["inputText"]

    @property
    def session_id(self) -> str:
        return self["sessionId"]

    @property
    def action_group(self) -> str:
        return self["actionGroup"]

    @property
    def api_path(self) -> str:
        return self["apiPath"]

    @property
    def http_method(self) -> str:
        return self["httpMethod"]

    @property
    def parameters(self) -> Optional[List[BedrockAgentProperty]]:
        return [BedrockAgentProperty(x) for x in self["parameters"]] if self.get("parameters") else None

    @property
    def request_body(self) -> Optional[BedrockAgentRequestBody]:
        return BedrockAgentRequestBody(self["requestBody"]) if self.get("requestBody") else None

    @property
    def agent(self) -> BedrockAgentInfo:
        return BedrockAgentInfo(self["agent"])

    @property
    def session_attributes(self) -> Dict[str, str]:
        return self["sessionAttributes"]

    @property
    def prompt_session_attributes(self) -> Dict[str, str]:
        return self["promptSessionAttributes"]

    # The following methods add compatibility with BaseProxyEvent
    @property
    def path(self) -> str:
        return self["apiPath"]

    @property
    def query_string_parameters(self) -> Optional[Dict[str, str]]:
        # In Bedrock Agent events, query string parameters are passed as undifferentiated parameters,
        # together with the other parameters. So we just return all parameters here.
        return {x["name"]: x["value"] for x in self["parameters"]} if self.get("parameters") else None

Ancestors

  • BaseProxyEvent
  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var action_group : str
Expand source code
@property
def action_group(self) -> str:
    return self["actionGroup"]
var agentBedrockAgentInfo
Expand source code
@property
def agent(self) -> BedrockAgentInfo:
    return BedrockAgentInfo(self["agent"])
var api_path : str
Expand source code
@property
def api_path(self) -> str:
    return self["apiPath"]
var input_text : str
Expand source code
@property
def input_text(self) -> str:
    return self["inputText"]
var message_version : str
Expand source code
@property
def message_version(self) -> str:
    return self["messageVersion"]
var parameters : Optional[List[BedrockAgentProperty]]
Expand source code
@property
def parameters(self) -> Optional[List[BedrockAgentProperty]]:
    return [BedrockAgentProperty(x) for x in self["parameters"]] if self.get("parameters") else None
var path : str
Expand source code
@property
def path(self) -> str:
    return self["apiPath"]
var prompt_session_attributes : Dict[str, str]
Expand source code
@property
def prompt_session_attributes(self) -> Dict[str, str]:
    return self["promptSessionAttributes"]
var query_string_parameters : Optional[Dict[str, str]]
Expand source code
@property
def query_string_parameters(self) -> Optional[Dict[str, str]]:
    # In Bedrock Agent events, query string parameters are passed as undifferentiated parameters,
    # together with the other parameters. So we just return all parameters here.
    return {x["name"]: x["value"] for x in self["parameters"]} if self.get("parameters") else None
var request_body : Optional[BedrockAgentRequestBody]
Expand source code
@property
def request_body(self) -> Optional[BedrockAgentRequestBody]:
    return BedrockAgentRequestBody(self["requestBody"]) if self.get("requestBody") else None
var session_attributes : Dict[str, str]
Expand source code
@property
def session_attributes(self) -> Dict[str, str]:
    return self["sessionAttributes"]
var session_id : str
Expand source code
@property
def session_id(self) -> str:
    return self["sessionId"]

Inherited members

class BedrockAgentInfo (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code
class BedrockAgentInfo(DictWrapper):
    @property
    def name(self) -> str:
        return self["name"]

    @property
    def id(self) -> str:  # noqa: A003
        return self["id"]

    @property
    def alias(self) -> str:
        return self["alias"]

    @property
    def version(self) -> str:
        return self["version"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var alias : str
Expand source code
@property
def alias(self) -> str:
    return self["alias"]
var id : str
Expand source code
@property
def id(self) -> str:  # noqa: A003
    return self["id"]
var name : str
Expand source code
@property
def name(self) -> str:
    return self["name"]
var version : str
Expand source code
@property
def version(self) -> str:
    return self["version"]

Inherited members

class BedrockAgentProperty (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code
class BedrockAgentProperty(DictWrapper):
    @property
    def name(self) -> str:
        return self["name"]

    @property
    def type(self) -> str:  # noqa: A003
        return self["type"]

    @property
    def value(self) -> str:
        return self["value"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var name : str
Expand source code
@property
def name(self) -> str:
    return self["name"]
var type : str
Expand source code
@property
def type(self) -> str:  # noqa: A003
    return self["type"]
var value : str
Expand source code
@property
def value(self) -> str:
    return self["value"]

Inherited members

class BedrockAgentRequestBody (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code
class BedrockAgentRequestBody(DictWrapper):
    @property
    def content(self) -> Dict[str, BedrockAgentRequestMedia]:
        return {k: BedrockAgentRequestMedia(v) for k, v in self["content"].items()}

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var content : Dict[str, BedrockAgentRequestMedia]
Expand source code
@property
def content(self) -> Dict[str, BedrockAgentRequestMedia]:
    return {k: BedrockAgentRequestMedia(v) for k, v in self["content"].items()}

Inherited members

class BedrockAgentRequestMedia (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code
class BedrockAgentRequestMedia(DictWrapper):
    @property
    def properties(self) -> List[BedrockAgentProperty]:
        return [BedrockAgentProperty(x) for x in self["properties"]]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var properties : List[BedrockAgentProperty]
Expand source code
@property
def properties(self) -> List[BedrockAgentProperty]:
    return [BedrockAgentProperty(x) for x in self["properties"]]

Inherited members