mypy-boto3-bedrock-agent-runtime

import boto3 from typing import TYPE_CHECKING, Dict, Any import os # These are for type checking only, not actual runtime imports if TYPE_CHECKING: from mypy_boto3_bedrock_agent_runtime.client import BedrockAgentRuntimeClient from mypy_boto3_bedrock_agent_runtime.type_defs import InvokeAgentRequestRequestTypeDef, InvokeAgentResponseTypeDef def get_bedrock_agent_runtime_client() -> "BedrockAgentRuntimeClient": """ Returns a type-hinted Bedrock Agent Runtime client. """ # In a real application, consider using environment variables or a config file # for region and credentials. # For this example, we'll assume region is set via AWS_DEFAULT_REGION or similar. return boto3.client("bedrock-agent-runtime") def invoke_agent_example(): client: BedrockAgentRuntimeClient = get_bedrock_agent_runtime_client() # Replace with your actual Agent ID and Alias ID # It's recommended to set these as environment variables (e.g., BEDROCK_AGENT_ID) agent_id = os.environ.get("BEDROCK_AGENT_ID", "YOUR_AGENT_ID") agent_alias_id = os.environ.get("BEDROCK_AGENT_ALIAS_ID", "YOUR_AGENT_ALIAS_ID") session_id = "test-session-123" # A unique identifier for the session input_text = "What is the weather like in Seattle?" if agent_id == "YOUR_AGENT_ID" or agent_alias_id == "YOUR_AGENT_ALIAS_ID": print("Please set BEDROCK_AGENT_ID and BEDROCK_AGENT_ALIAS_ID environment variables to run this example.") print("You can find these in the AWS Bedrock Agents console.") return request_params: InvokeAgentRequestRequestTypeDef = { "agentId": agent_id, "agentAliasId": agent_alias_id, "sessionId": session_id, "inputText": input_text, "enableTrace": True, # Set to False for production to reduce logging } print(f"Invoking agent with input: '{input_text}'") try: response: InvokeAgentResponseTypeDef = client.invoke_agent(**request_params) # Process the response stream response_body = response.get("completion") if response_body: print("Agent response:") for event in response_body: if "chunk" in event: chunk = event["chunk"] text = chunk.get("bytes").decode("utf-8") print(f" {text}", end="") print("\n") # Newline after all chunks else: print("No completion received.") except client.exceptions.ResourceNotFoundException as e: print(f"Error: Agent or Alias not found. Ensure 'BEDROCK_AGENT_ID' and 'BEDROCK_AGENT_ALIAS_ID' are correct. {e}") except client.exceptions.ValidationException as e: print(f"Validation Error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") if __name__ == "__main__": invoke_agent_example()