┌─────────────────────────────────────────────────────────────┐
│                     Application Layer                        │
│  (Your code using bedrock-agent-runtime.invoke_agent)       │
└───────────────────────────┬─────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────┐
│                    Bedrock Agent Runtime                     │
│  - Session management                                        │
│  - Streaming responses                                       │
│  - Trace/reasoning visibility                                │
└───────┬──────────────────┬──────────────────┬───────────────┘
        │                  │                  │
┌───────▼────────┐  ┌─────▼──────┐  ┌────────▼──────────┐
│ Foundation     │  │   Action   │  │    Knowledge      │
│ Model          │  │   Groups   │  │    Bases          │
│ (Claude, etc.) │  │  (Lambda,  │  │  (Vector DB +     │
│                │  │   OpenAPI) │  │   Documents)      │
└────────────────┘  └────────────┘  └───────────────────┘

import boto3
import json

bedrock_agent = boto3.client('bedrock-agent', region_name='us-east-1')

def create_agent(
    agent_name: str,
    foundation_model: str = "anthropic.claude-3-5-sonnet-20241022-v2:0",
    instructions: str = None,
    description: str = None,
    idle_session_ttl: int = 600
) -> dict:
    """
    Create a Bedrock Agent.

    Args:
        agent_name: Name of the agent
        foundation_model: Model ID (Claude 3.5 Sonnet, Haiku, etc.)
        instructions: System prompt defining agent behavior
        description: Human-readable description
        idle_session_ttl: Session timeout in seconds (default 600)

    Returns:
        Agent details including agentId, agentArn, agentStatus
    """

    # Default instructions if none provided
    if instructions is None:
        instructions = """You are a helpful AI assistant.
        Follow user requests carefully and provide accurate, helpful responses.
        When you need to use tools or access knowledge, do so to provide better answers."""

    try:
        response = bedrock_agent.create_agent(
            agentName=agent_name,
            foundationModel=foundation_model,
            instruction=instructions,
            description=description or f"Bedrock Agent: {agent_name}",
            idleSessionTTLInSeconds=idle_session_ttl,
            # Optional: Add tags
            tags={
                'Environment': 'production',
                'ManagedBy': 'bedrock-agents-skill'
            }
        )

        agent = response['agent']
        print(f"✓ Created agent: {agent['agentName']}")
        print(f"  Agent ID: {agent['agentId']}")
        print(f"  Status: {agent['agentStatus']}")
        print(f"  Foundation Model: {agent['foundationModel']}")

        return agent

    except Exception as e:
        print(f"✗ Failed to create agent: {str(e)}")
        raise

Your responsibilities:
- Answer customer questions about orders, products, and policies
- Look up order status using available tools
- Provide accurate information from the knowledge base
- Be helpful, professional, and empathetic

Always verify information before providing answers.
If you cannot help, escalate to a human agent.""",
description="AI-powered customer support agent",
idle_session_ttl=1800  # 30 minutes

**Foundation Models**:
- `anthropic.claude-3-5-sonnet-20241022-v2:0` - Best reasoning
- `anthropic.claude-3-5-haiku-20241022-v1:0` - Fast responses
- `anthropic.claude-3-opus-20240229-v1:0` - Maximum capability
- `amazon.titan-text-premier-v1:0` - AWS native model

def prepare_agent(agent_id: str) -> dict:
    """
    Prepare agent for use. Required after creation or configuration changes.

    Args:
        agent_id: ID of the agent to prepare

    Returns:
        Prepared agent details
    """

    try:
        response = bedrock_agent.prepare_agent(agentId=agent_id)

        print(f"✓ Agent preparation started")
        print(f"  Agent ID: {agent_id}")
        print(f"  Status: {response['agentStatus']}")
        print(f"  Prepared At: {response['preparedAt']}")

        return response

    except Exception as e:
        print(f"✗ Failed to prepare agent: {str(e)}")
        raise

start_time = time.time()
while time.time() - start_time < max_wait:
    response = bedrock_agent.get_agent(agentId=agent_id)
    status = response['agent']['agentStatus']

    if status in ['PREPARED', 'VERSIONED']:
        print(f"✓ Agent ready (status: {status})")
        return True
    elif status == 'FAILED':
        print(f"✗ Agent preparation failed")
        return False

    print(f"  Waiting for agent... (status: {status})")
    time.sleep(5)

print(f"✗ Timeout waiting for agent")
return False

undefined

def create_lambda_action_group(
    agent_id: str,
    action_group_name: str,
    lambda_arn: str,
    description: str,
    api_schema: dict
) -> dict:
    """
    Create an action group that invokes Lambda functions.

    Args:
        agent_id: ID of the agent
        action_group_name: Name of the action group
        lambda_arn: ARN of the Lambda function to invoke
        description: Description of what this action group does
        api_schema: OpenAPI schema defining available operations

    Returns:
        Action group details
    """

    try:
        response = bedrock_agent.create_agent_action_group(
            agentId=agent_id,
            agentVersion='DRAFT',
            actionGroupName=action_group_name,
            description=description,
            actionGroupExecutor={
                'lambda': lambda_arn
            },
            apiSchema={
                'payload': json.dumps(api_schema)
            },
            actionGroupState='ENABLED'
        )

        action_group = response['agentActionGroup']
        print(f"✓ Created action group: {action_group['actionGroupName']}")
        print(f"  Action Group ID: {action_group['actionGroupId']}")
        print(f"  Lambda: {lambda_arn}")

        return action_group

    except Exception as e:
        print(f"✗ Failed to create action group: {str(e)}")
        raise

**OpenAPI Action Group** (External REST API):

```python
def create_openapi_action_group(
    agent_id: str,
    action_group_name: str,
    api_schema: dict,
    description: str
) -> dict:
    """
    Create an action group for external REST APIs.
    Agent will make HTTP requests to the API defined in the schema.
    """

    try:
        response = bedrock_agent.create_agent_action_group(
            agentId=agent_id,
            agentVersion='DRAFT',
            actionGroupName=action_group_name,
            description=description,
            actionGroupExecutor={
                'customControl': 'RETURN_CONTROL'  # Return to application for execution
            },
            apiSchema={
                'payload': json.dumps(api_schema)
            },
            actionGroupState='ENABLED'
        )

        return response['agentActionGroup']

    except Exception as e:
        print(f"✗ Failed to create OpenAPI action group: {str(e)}")
        raise

def associate_knowledge_base(
    agent_id: str,
    knowledge_base_id: str,
    description: str = "Agent knowledge base"
) -> dict:
    """
    Associate a knowledge base with an agent for RAG.

    Args:
        agent_id: ID of the agent
        knowledge_base_id: ID of the knowledge base to associate
        description: Description of the knowledge base purpose

    Returns:
        Association details
    """

    try:
        response = bedrock_agent.associate_agent_knowledge_base(
            agentId=agent_id,
            agentVersion='DRAFT',
            knowledgeBaseId=knowledge_base_id,
            description=description,
            knowledgeBaseState='ENABLED'
        )

        kb = response['agentKnowledgeBase']
        print(f"✓ Associated knowledge base")
        print(f"  Knowledge Base ID: {kb['knowledgeBaseId']}")
        print(f"  Description: {kb['description']}")

        return kb

    except Exception as e:
        print(f"✗ Failed to associate knowledge base: {str(e)}")
        raise

undefined

def create_agent_alias(
    agent_id: str,
    alias_name: str,
    description: str = None
) -> dict:
    """
    Create an agent alias for deployment.

    Args:
        agent_id: ID of the agent
        alias_name: Name for the alias (e.g., 'production', 'staging')
        description: Description of the alias

    Returns:
        Alias details including agentAliasId
    """

    try:
        response = bedrock_agent.create_agent_alias(
            agentId=agent_id,
            agentAliasName=alias_name,
            description=description or f"{alias_name} environment"
        )

        alias = response['agentAlias']
        print(f"✓ Created alias: {alias['agentAliasName']}")
        print(f"  Alias ID: {alias['agentAliasId']}")
        print(f"  Status: {alias['agentAliasStatus']}")

        return alias

    except Exception as e:
        print(f"✗ Failed to create alias: {str(e)}")
        raise

start_time = time.time()
while time.time() - start_time < max_wait:
    response = bedrock_agent.get_agent_alias(
        agentId=agent_id,
        agentAliasId=alias_id
    )
    status = response['agentAlias']['agentAliasStatus']

    if status == 'PREPARED':
        print(f"✓ Alias ready")
        return True
    elif status == 'FAILED':
        print(f"✗ Alias preparation failed")
        return False

    print(f"  Waiting for alias... (status: {status})")
    time.sleep(5)

return False

undefined

import boto3

bedrock_agent_runtime = boto3.client('bedrock-agent-runtime', region_name='us-east-1')

def invoke_agent(
    agent_id: str,
    agent_alias_id: str,
    session_id: str,
    input_text: str,
    enable_trace: bool = True
) -> dict:
    """
    Invoke a Bedrock Agent with streaming response.

    Args:
        agent_id: ID of the agent
        agent_alias_id: ID of the agent alias
        session_id: Unique session ID for conversation tracking
        input_text: User's input text
        enable_trace: Enable trace for reasoning visibility

    Returns:
        Complete response with text, traces, and citations
    """

    try:
        response = bedrock_agent_runtime.invoke_agent(
            agentId=agent_id,
            agentAliasId=agent_alias_id,
            sessionId=session_id,
            inputText=input_text,
            enableTrace=enable_trace
        )

        # Process streaming response
        result = {
            'completion': '',
            'traces': [],
            'citations': []
        }

        event_stream = response['completion']

        for event in event_stream:
            if 'chunk' in event:
                chunk = event['chunk']
                if 'bytes' in chunk:
                    text = chunk['bytes'].decode('utf-8')
                    result['completion'] += text
                    print(text, end='', flush=True)

            elif 'trace' in event:
                trace = event['trace']['trace']
                result['traces'].append(trace)

                # Print reasoning steps
                if enable_trace:
                    if 'orchestrationTrace' in trace:
                        orch = trace['orchestrationTrace']
                        if 'rationale' in orch:
                            print(f"\n[Reasoning] {orch['rationale']['text']}")
                        if 'invocationInput' in orch:
                            print(f"[Action] {orch['invocationInput']}")

            elif 'returnControl' in event:
                # Handle return of control for action group execution
                result['return_control'] = event['returnControl']

        print("\n")
        return result

    except Exception as e:
        print(f"✗ Failed to invoke agent: {str(e)}")
        raise

**Multi-Turn Conversation**:

```python
def chat_with_agent(
    agent_id: str,
    agent_alias_id: str,
    enable_trace: bool = False
):
    """
    Interactive chat session with the agent.
    Maintains conversation context across turns.
    """

    session_id = str(uuid.uuid4())
    print(f"Chat session started (ID: {session_id})")
    print("Type 'exit' to end the conversation\n")

    while True:
        user_input = input("You: ").strip()

        if user_input.lower() in ['exit', 'quit', 'bye']:
            print("Ending conversation...")
            break

        if not user_input:
            continue

        print("\nAgent: ", end='', flush=True)

        response = invoke_agent(
            agent_id=agent_id,
            agent_alias_id=agent_alias_id,
            session_id=session_id,
            input_text=user_input,
            enable_trace=enable_trace
        )

        print()  # New line after response

**Invoke with Session State**:

```python
def invoke_agent_with_state(
    agent_id: str,
    agent_alias_id: str,
    session_id: str,
    input_text: str,
    session_state: dict = None
) -> dict:
    """
    Invoke agent with custom session state.

    Session state can include:
    - promptSessionAttributes: Dynamic variables for prompts
    - sessionAttributes: Persistent session data
    """

    invoke_params = {
        'agentId': agent_id,
        'agentAliasId': agent_alias_id,
        'sessionId': session_id,
        'inputText': input_text
    }

    if session_state:
        invoke_params['sessionState'] = session_state

    response = bedrock_agent_runtime.invoke_agent(**invoke_params)

    # Process response...
    result = {'completion': ''}
    for event in response['completion']:
        if 'chunk' in event and 'bytes' in event['chunk']:
            result['completion'] += event['chunk']['bytes'].decode('utf-8')

    return result

undefined

def delete_agent_memory(
    agent_id: str,
    agent_alias_id: str,
    session_id: str
) -> dict:
    """
    Delete conversation memory for a specific session.

    Args:
        agent_id: ID of the agent
        agent_alias_id: ID of the agent alias
        session_id: Session ID to clear

    Returns:
        Deletion confirmation
    """

    try:
        response = bedrock_agent_runtime.delete_agent_memory(
            agentId=agent_id,
            agentAliasId=agent_alias_id,
            memoryId=session_id
        )

        print(f"✓ Deleted agent memory for session: {session_id}")
        return response

    except Exception as e:
        print(f"✗ Failed to delete memory: {str(e)}")
        raise

undefined

def update_agent(
    agent_id: str,
    agent_name: str = None,
    instructions: str = None,
    foundation_model: str = None,
    description: str = None
) -> dict:
    """
    Update agent configuration.

    After update, must prepare agent again before use.
    """

    try:
        # Get current agent details
        current = bedrock_agent.get_agent(agentId=agent_id)['agent']

        update_params = {
            'agentId': agent_id,
            'agentName': agent_name or current['agentName'],
            'foundationModel': foundation_model or current['foundationModel'],
            'instruction': instructions or current.get('instruction', '')
        }

        if description:
            update_params['description'] = description

        response = bedrock_agent.update_agent(**update_params)

        print(f"✓ Updated agent: {response['agent']['agentName']}")
        print("  Note: Prepare agent before use")

        return response['agent']

    except Exception as e:
        print(f"✗ Failed to update agent: {str(e)}")
        raise

New capabilities:
- Proactively suggest related products
- Offer discounts for service issues
- Escalate complex cases to supervisors

Maintain professional, empathetic tone."""

undefined

def list_agents() -> list:
    """List all agents in the account."""

    try:
        response = bedrock_agent.list_agents(maxResults=50)

        agents = response.get('agentSummaries', [])

        print(f"Found {len(agents)} agents:")
        for agent in agents:
            print(f"\n  Name: {agent['agentName']}")
            print(f"  ID: {agent['agentId']}")
            print(f"  Status: {agent['agentStatus']}")
            print(f"  Updated: {agent['updatedAt']}")

        return agents

    except Exception as e:
        print(f"✗ Failed to list agents: {str(e)}")
        raise

def delete_agent(agent_id: str, skip_resource_in_use_check: bool = False) -> dict:
    """
    Delete an agent and all associated resources.

    Args:
        agent_id: ID of the agent to delete
        skip_resource_in_use_check: Skip check for aliases
    """

    try:
        response = bedrock_agent.delete_agent(
            agentId=agent_id,
            skipResourceInUseCheck=skip_resource_in_use_check
        )

        print(f"✓ Deleted agent: {agent_id}")
        print(f"  Status: {response['agentStatus']}")

        return response

    except Exception as e:
        print(f"✗ Failed to delete agent: {str(e)}")
        raise

undefined

def single_turn_query(agent_id: str, agent_alias_id: str, query: str) -> str:
    """
    Execute a single query without conversation context.
    Each invocation is independent.
    """

    # Use unique session ID for each request
    session_id = str(uuid.uuid4())

    response = bedrock_agent_runtime.invoke_agent(
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        inputText=query,
        enableTrace=False
    )

    completion = ''
    for event in response['completion']:
        if 'chunk' in event and 'bytes' in event['chunk']:
            completion += event['chunk']['bytes'].decode('utf-8')

    return completion

undefined

class ConversationSession:
    """Manage a multi-turn conversation with context."""

    def __init__(self, agent_id: str, agent_alias_id: str):
        self.agent_id = agent_id
        self.agent_alias_id = agent_alias_id
        self.session_id = str(uuid.uuid4())
        self.history = []

    def send_message(self, message: str) -> str:
        """Send a message and get response."""

        response = bedrock_agent_runtime.invoke_agent(
            agentId=self.agent_id,
            agentAliasId=self.agent_alias_id,
            sessionId=self.session_id,
            inputText=message
        )

        completion = ''
        for event in response['completion']:
            if 'chunk' in event and 'bytes' in event['chunk']:
                completion += event['chunk']['bytes'].decode('utf-8')

        self.history.append({'user': message, 'agent': completion})
        return completion

    def reset(self):
        """Clear conversation memory."""
        bedrock_agent_runtime.delete_agent_memory(
            agentId=self.agent_id,
            agentAliasId=self.agent_alias_id,
            memoryId=self.session_id
        )
        self.history = []

undefined

def rag_query(agent_id: str, agent_alias_id: str, query: str) -> dict:
    """
    Query agent with RAG capabilities.
    Returns response with citations from knowledge base.
    """

    session_id = str(uuid.uuid4())

    response = bedrock_agent_runtime.invoke_agent(
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        inputText=query,
        enableTrace=True
    )

    result = {
        'answer': '',
        'citations': [],
        'retrieved_docs': []
    }

    for event in response['completion']:
        if 'chunk' in event and 'bytes' in event['chunk']:
            result['answer'] += event['chunk']['bytes'].decode('utf-8')

        elif 'trace' in event:
            trace = event['trace']['trace']

            # Extract knowledge base retrieval
            if 'orchestrationTrace' in trace:
                orch = trace['orchestrationTrace']
                if 'observation' in orch:
                    obs = orch['observation']
                    if 'knowledgeBaseLookupOutput' in obs:
                        kb_output = obs['knowledgeBaseLookupOutput']
                        result['retrieved_docs'].extend(
                            kb_output.get('retrievedReferences', [])
                        )

    return result

undefined

def tool_use_example(agent_id: str, agent_alias_id: str, request: str):
    """
    Agent uses action groups to invoke tools.
    Demonstrates tool invocation flow.
    """

    session_id = str(uuid.uuid4())

    response = bedrock_agent_runtime.invoke_agent(
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        inputText=request,
        enableTrace=True
    )

    for event in response['completion']:
        if 'trace' in event:
            trace = event['trace']['trace']

            if 'orchestrationTrace' in trace:
                orch = trace['orchestrationTrace']

                # Agent planning
                if 'rationale' in orch:
                    print(f"[Planning] {orch['rationale']['text']}")

                # Agent invoking tool
                if 'invocationInput' in orch:
                    inv = orch['invocationInput']
                    if 'actionGroupInvocationInput' in inv:
                        action = inv['actionGroupInvocationInput']
                        print(f"[Tool Call] {action['actionGroupName']}")
                        print(f"  API: {action.get('apiPath', 'N/A')}")
                        print(f"  Parameters: {action.get('parameters', {})}")

                # Tool result
                if 'observation' in orch:
                    obs = orch['observation']
                    if 'actionGroupInvocationOutput' in obs:
                        output = obs['actionGroupInvocationOutput']
                        print(f"[Tool Result] {output.get('text', '')}")

        elif 'chunk' in event and 'bytes' in event['chunk']:
            text = event['chunk']['bytes'].decode('utf-8')
            print(text, end='', flush=True)

undefined

good_instructions = """You are a financial advisor agent for retail banking customers.

Your capabilities:
- Answer questions about savings accounts, checking accounts, and credit cards
- Look up account balances and transaction history using available tools
- Provide information about loan products from the knowledge base
- Help customers understand fees and policies

Your guidelines:
- Always verify customer identity before accessing account information
- Provide accurate information from official sources only
- If you cannot help with something, explain why and suggest alternatives
- Use professional but friendly language
- Cite sources when providing policy or fee information

Your limitations:
- You cannot make transactions or transfer money
- You cannot change account settings
- You cannot access accounts without proper verification
- For complex issues, escalate to a human advisor

Session context:
- Customer tier: {{customer_tier}}
- Customer since: {{customer_since}}
- Primary account: {{primary_account_type}}
"""

undefined

undefined

def invoke_agent_safe(agent_id: str, agent_alias_id: str, session_id: str, input_text: str):
    """Invoke agent with comprehensive error handling."""

    try:
        response = bedrock_agent_runtime.invoke_agent(
            agentId=agent_id,
            agentAliasId=agent_alias_id,
            sessionId=session_id,
            inputText=input_text
        )

        completion = ''

        for event in response['completion']:
            try:
                if 'chunk' in event and 'bytes' in event['chunk']:
                    completion += event['chunk']['bytes'].decode('utf-8')

                elif 'internalServerException' in event:
                    raise Exception("Internal server error")

                elif 'validationException' in event:
                    raise ValueError("Validation error")

                elif 'throttlingException' in event:
                    raise Exception("Rate limit exceeded")

            except Exception as e:
                print(f"Stream error: {str(e)}")
                continue

        return completion

    except bedrock_agent_runtime.exceptions.ResourceNotFoundException:
        print("Agent or alias not found")
        raise
    except bedrock_agent_runtime.exceptions.AccessDeniedException:
        print("Access denied - check IAM permissions")
        raise
    except Exception as e:
        print(f"Invocation error: {str(e)}")
        raise