삼성전자 로봇 설비 데이터 인사이트 혁신, Part 2: Amazon Bedrock AgentCore와 MCP를 활용한 AI 챗봇

# Starter Toolkit 설치
$ pip install bedrock-agentcore-starter-toolkit 

# Agent 설정
$ agentcore configure -e my_agent.py 

# Agent 배포
$ agentcore launch 

# Agent 호출 테스트
$ agentcore invoke '{"prompt": "지난주 오류 발생 건수는?"}'

# my_agent.py

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel

app = BedrockAgentCoreApp()
model = BedrockModel(model_id="global.anthropic.claude-sonnet-4-5-20250929-v1:0")

system_prompt = """
You are an intelligent assistant that helps users analyze kitchen robot facility data.
... (생략) ...
"""

@app.entrypoint
async def strands_agent_bedrock(payload):
    user_input = payload.get("prompt")
    
    # MCP 클라이언트 및 도구 초기화
    chart_client = create_mcp_client()
    chart_tools = get_full_tools_list(chart_client)
    
    # Agent 생성
    agent = Agent(
        model=model,
        tools=[
            gen_query,    # SQL 쿼리 생성
            exec_query,   # SQL 쿼리 실행
            chart_tools   # 차트 생성 도구
        ],
        system_prompt=system_prompt
    )
    
    # 스트리밍 응답 반환
    stream = agent.stream_async(user_input)
    async for event in stream:
        if "data" in event:
            yield {"type": "text", "content": event['data']}

if __name__ == "__main__":
    app.run()

from strands import tool, Agent 
from strands.models import BedrockModel 

@tool
async def gen_query(input: str) -> str: 
    """
    Generate SQL Query for kitchen robot operational data from user input.  
    Returns the generated SQL query as a formatted string.  
    
    Args:  input (str): User input to generate SQL for kitchen robot logs 
    """ 
    
    model = BedrockModel(model_id="global.anthropic.claude-sonnet-4-5-20250929-v1:0") 
    system_prompt = f"""
    You are an AWS Athena SQL query generator for kitchen robot operational data. 
    Generate optimized Presto SQL queries for AWS Athena.  
    
    ## Database Environment  
    - Engine: AWS Athena (Presto SQL)  
    - Data Format: Parquet files in S3  
    - Cost: Charged by data scanned, so optimization is critical  
    
    ## Current Database Schema  
    Table Name: robot_logs.operational_data  
    Column Descriptions:  
        "timestamp" (TIMESTAMP): 데이터의 기준 시간  
        "id" (STRING): 설비 구분 ID  
        "menu_name" (STRING): 소분할 메뉴 이름  
        "facility_status" (INT): 설비 동작 상태
        ... (생략) ...  
    
    ## Athena Query Optimization Rules  
    1. ALWAYS use column names instead of SELECT *  
    2. ALWAYS add LIMIT clause (default 100)  
    3. Use proper Presto SQL syntax with semicolon  
    4. Return only the SQL query, no explanations  
    
    ## Performance Best Practices  
    - Minimize columns in SELECT to reduce scan costs  
    - Use DATE functions for timestamp comparisons  
    - Push down predicates early in WHERE clause  
    - Avoid SELECT * to minimize costs  
    
    Convert the user input to a SQL query:
    """ 
    
    # 별도의 Agent를 사용하여 SQL 생성
    agent = Agent( 
        model=model, 
        system_prompt=system_prompt, 
    ) 
    res = await agent.invoke_async(input) 
    sql_query = res.message.content if hasattr(res.message, 'content') else str(res.message)
    
    # 포맷팅하여 반환 - 사용자에게 명확하게 표시 
    return f"```sql\n{sql_query}\n```"

## Database Environment
- Engine: AWS Athena (Presto SQL)
- Data Format: Parquet files in S3 
- Cost: Charged by data scanned, so optimization is critical

## Current Database Schema
Table Name: robot_logs.operational_data 
Column Descriptions: 
    "timestamp" (TIMESTAMP): 데이터의 기준 시간  
    "id" (STRING): 설비 구분 ID  
    "menu_name" (STRING): 소분할 메뉴 이름  
    "facility_status" (INT): 설비 동작 상태
    <이하 생략>

## Athena Query Optimization Rules 
1. ALWAYS use column names instead of SELECT *
2. ALWAYS add LIMIT clause (default 100)
3. Use proper Presto SQL syntax with semicolon 
4. Return only the SQL query, no explanations

## Example Queries
Input: "8월 3일에 제공된 마지막 메뉴는?"
Output: SELECT menu_name 
        FROM robot_logs.operational_data 
        WHERE timestamp >= CAST('2025-08-03 00:00:00' AS TIMESTAMP)
          AND timestamp < CAST('2025-08-04 00:00:00' AS TIMESTAMP)
        ORDER BY timestamp DESC 
        LIMIT 1;

from typing import List, Any
from pyathena import connect
from pyathena.cursor import DictCursor
from strands import tool

@tool
def exec_query(query: str) -> List[dict[str, Any]]:
    """
    Execute SQL queries against AWS Athena.

    Args:
        query (str): Valid Presto SQL SELECT statement
    """
    # PyAthena 연결 생성
    conn = connect(
        s3_staging_dir="s3://my-bucket/athena-results/",
        region_name="us-east-1",
        work_group="primary",
        cursor_class=DictCursor  # Returns results as dictionaries
    )
    
    cursor = conn.cursor()
    try:
        cursor.execute(query)
        return cursor.fetchall()
    except Exception as e:
        print(f"Query execution error: {e}")
        raise

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
import { Context, Handler } from "aws-lambda";
import { randomUUID } from "crypto";

interface SaveChartEvent {
  readonly title: string;
  readonly data: Record<string, any>[];
}

/*
MCP Tool Schema
[
  {
    "description": "차트 데이터를 데이터베이스에 저장하는 도구",
    "inputSchema": {
      "properties": {
        "data": {
          "description": "차트 데이터 배열 - 각 항목은 키-값 쌍의 객체",
          "items": {
            "type": "object"
          },
          "type": "array"
        },
        "title": {
          "description": "저장할 차트의 제목",
          "type": "string"
        }
      },
      "required": [
        "title",
        "data"
      ],
      "type": "object"
    },
    "name": "save_chart"
  }
]
*/
export const handler: Handler = async (
  event: SaveChartEvent,
  context: Context,
) => {
  console.log("event", event);
  console.log("context", context);

  const tableName = process.env.DYNAMODB_TABLE;

  const client = new DynamoDBClient({});
  const docClient = DynamoDBDocumentClient.from(client);

  const uid = randomUUID();

  await docClient.send(
    new PutCommand({
      TableName: tableName,
      Item: {
        PK: "DEFAULT",
        SK: `CHART#${uid}`,
        data: event.data,
        title: event.title,
        timestamp: Date.now(),
      },
    }),
  );

  return "The data is successfully saved.";
};

import boto3

gateway_client = boto3.client(
    'bedrock-agentcore-control', 
    region_name='your aws region'
)

auth_config = {
    "customJWTAuthorizer": { 
        "allowedClients": ['your cognito m2m client'], 
        "discoveryUrl": 'your cognito discovery url' 
    }
}

gateway_response = gateway_client.create_gateway(
    name="mcp-chart",
    description="Mcp Chart API Gateway",
    roleArn='your agentcore role arn',
    protocolType='MCP',
    authorizerType='CUSTOM_JWT',
    authorizerConfiguration=auth_config,
)

lambda_target_config = {
    "mcp": {
        "lambda": {
            "lambdaArn": 'your save chart lambda function arn',
            "toolSchema": {
                "inlinePayload": [
                    {
                        "name": "save_chart",
                        "description": "tool for saving chart data to database",
                        "inputSchema": {
                            "type": "object",
                            "properties": {
                                "data": {
                                    "type": "array",
                                    "description": "Array of char data - each item is an object with key-value pairs",
                                    "items": {
                                        "type": "object"
                                    }
                                },
                                "title": {
                                    "type": "string",
                                    "description": "Title of the chart to save"
                                }
                            },
                            "required": ["title", "data"]
                        }
                    }
                ]
            }
        }
    }
}

credential_config = [ 
    {
        "credentialProviderType" : "GATEWAY_IAM_ROLE"
    }
]

targetname='ChartLambdaFunctions'


target_response = gateway_client.create_gateway_target(
    gatewayIdentifier=gateway_response['gatewayId'],
    name=targetname,
    description='Chart Lambda Target',
    targetConfiguration=lambda_target_config,
    credentialProviderConfigurations=credential_config)

from typing import List

import requests
from mcp.client.streamable_http import streamablehttp_client
from strands.tools.mcp.mcp_client import MCPClient, MCPAgentTool

# Edit Variable for you environments
CLIENT_ID = 'your cognito client id'
CLIENT_SECRET = 'your cognito client secret'
COGNITO_DOMAIN = 'your cognito domain'
MCP_ENDPOINT = 'your agentcore gateway endpoint'


def get_token_url() -> str:
    return f"{COGNITO_DOMAIN}/oauth2/token"


def fetch_access_token(client_id: str, client_secret: str, token_url: str) -> str:
    response = requests.post(
        token_url,
        data="grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}".format(
            client_id=client_id,
            client_secret=client_secret,
        ),
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )
    return response.json()['access_token']


def create_mcp_client() -> MCPClient:
    access_token = fetch_access_token(CLIENT_ID, CLIENT_SECRET, get_token_url())
    mcp_client = MCPClient(
        lambda: streamablehttp_client(MCP_ENDPOINT, headers={"Authorization": f"Bearer {access_token}"}))
    return mcp_client


def get_full_tools_list(client: MCPClient) -> List[MCPAgentTool]:
    """
    List tools w/ support for pagination
    """
    more_tools = True
    tools = []
    pagination_token = None

    while more_tools:
        tmp_tools = client.list_tools_sync(pagination_token=pagination_token)
        tools.extend(tmp_tools)
        if tmp_tools.pagination_token is None:
            more_tools = False
        else:
            more_tools = True
            pagination_token = tmp_tools.pagination_token

    return tools

import {BedrockAgentCoreClient, InvokeAgentRuntimeCommand} from "@aws-sdk/client-bedrock-agentcore";

export interface ChatMessageInput {
    sessionId: string;
    message: string;
    agentRuntimeArn: string;
}

export async function createChatStream(input: ChatMessageInput) {
    const client = new BedrockAgentCoreClient();
    const cmd = new InvokeAgentRuntimeCommand({
        agentRuntimeArn: input.agentRuntimeArn,
        contentType: 'application/json',
        accept: 'application/json',
        payload: JSON.stringify({
            session_id: input.sessionId,
            prompt: input.message,
        })
    });

    const output = await client.send(cmd);
    const sourceStream = output.response?.transformToWebStream();
    
    if (!sourceStream) {
        throw new Error('No response stream available');
    }

    return new ReadableStream({
        async start(controller) {
            try {
                const reader = sourceStream.getReader();
                
                while (true) {
                    const { done, value } = await reader.read();
                    
                    if (done) {
                        controller.close();
                        break;
                    }
                    
                    controller.enqueue(value);
                }
            } catch (error) {
                const serializableError = new Error(
                    error instanceof Error ? error.message : 'Failed to invoke agent runtime'
                );
                if (error instanceof Error) {
                    serializableError.name = error.name;
                    serializableError.stack = error.stack;
                }
                controller.error(serializableError);
            }
        }
    });
}]