Lab 3: Architecture Deep Dive¶

import json
import boto3
from uuid import uuid4

bedrock_agentcore = boto3.client('bedrock-agent-runtime')

def lambda_handler(event, context):
    """
    Handles chat requests for AgentCore integration
    """
    # 1. Extract request parameters
    body = json.loads(event['body'])
    query = body.get('query', '')
    session_id = body.get('sessionId') or str(uuid4())
    memory_id = body.get('memoryId')  # For cross-session memory

    # 2. Invoke AgentCore Runtime
    try:
        response = bedrock_agentcore.invoke_agent(
            agentId=os.environ['AGENT_RUNTIME_ARN'],
            sessionId=session_id,
            memoryId=memory_id,
            inputText=query,
            enableTrace=True  # Enable detailed tracing
        )

        # 3. Process streaming response
        answer = ""
        citations = []
        tool_invocations = []

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

            if 'trace' in event:
                # Extract tool invocations and citations
                trace = event['trace']
                tool_invocations.append(extract_tool_info(trace))
                citations.extend(extract_citations(trace))

        # 4. Format and return response
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json',
                'Access-Control-Allow-Origin': '*'
            },
            'body': json.dumps({
                'answer': answer,
                'citations': citations,
                'toolInvocations': tool_invocations,
                'sessionId': session_id,
                'memoryId': memory_id
            })
        }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': 'AgentCore invocation failed',
                'message': str(e)
            })
        }

from fastapi import FastAPI
from mcp import MCPServer, Tool
import requests
import os

app = FastAPI()
mcp_server = MCPServer()

# Tool 1: Search Coveo
@mcp_server.tool(
    name="search_coveo",
    description="Search the Coveo knowledge base for relevant information",
    parameters={
        "query": {
            "type": "string",
            "description": "The search query"
        }
    }
)
async def search_coveo(query: str):
    """Search Coveo knowledge base"""
    coveo_request = {
        'q': query,
        'organizationId': os.environ['COVEO_ORG_ID'],
        'searchHub': 'workshop',
        'numberOfResults': 10
    }

    headers = {
        'Authorization': f'Bearer {os.environ["COVEO_API_KEY"]}',
        'Content-Type': 'application/json'
    }

    response = requests.post(
        'https://platform.cloud.coveo.com/rest/search/v2/',
        json=coveo_request,
        headers=headers
    )

    results = response.json()
    return {
        'results': results.get('results', []),
        'totalCount': results.get('totalCount', 0)
    }

# Tool 2: Passage Retrieval
@mcp_server.tool(
    name="passage_retrieval",
    description="Retrieve relevant text passages from Coveo knowledge base",
    parameters={
        "query": {
            "type": "string",
            "description": "The query to find relevant passages"
        }
    }
)
async def passage_retrieval(query: str):
    """Retrieve passages from Coveo"""
    coveo_request = {
        'q': query,
        'organizationId': os.environ['COVEO_ORG_ID'],
        'searchHub': 'workshop',
        'numberOfPassages': 5
    }

    headers = {
        'Authorization': f'Bearer {os.environ["COVEO_API_KEY"]}',
        'Content-Type': 'application/json'
    }

    response = requests.post(
        'https://platform.cloud.coveo.com/rest/search/v2/passages',
        json=coveo_request,
        headers=headers
    )

    passages = response.json()
    return {
        'passages': passages.get('passages', [])
    }

# Tool 3: Answer Question
@mcp_server.tool(
    name="answer_question",
    description="Generate an AI answer using Coveo Answer API",
    parameters={
        "query": {
            "type": "string",
            "description": "The question to answer"
        }
    }
)
async def answer_question(query: str):
    """Generate answer using Coveo Answer API"""
    coveo_request = {
        'q': query,
        'organizationId': os.environ['COVEO_ORG_ID'],
        'searchHub': 'workshop',
        'answerConfigurationId': os.environ['COVEO_ANSWER_CONFIG_ID']
    }

    headers = {
        'Authorization': f'Bearer {os.environ["COVEO_API_KEY"]}',
        'Content-Type': 'application/json'
    }

    response = requests.post(
        'https://platform.cloud.coveo.com/rest/search/v2/answer',
        json=coveo_request,
        headers=headers
    )

    answer_data = response.json()
    return {
        'answer': answer_data.get('answer', ''),
        'citations': answer_data.get('citations', [])
    }

# MCP Server endpoints
@app.post("/mcp/tools/list")
async def list_tools():
    """List available tools"""
    return mcp_server.list_tools()

@app.post("/mcp/tools/call")
async def call_tool(request: dict):
    """Call a specific tool"""
    return await mcp_server.call_tool(
        request['tool_name'],
        request['parameters']
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=3000)