여기에서는 AgentCore Runtime을 이용해 LangGraph, Strands SDK를 이용하여, MCP를 활용하는 Agent를 Production 환경에 배포하고 안전하게 사용하는 방법에 대해 설명합니다.
전체적인 Architecture는 아래와 같습니다. 여기서는 MCP를 지원하는 Strands와 LangGraph agent를 AgentCore를 이용해 배포하고 streamlit 애플리케이션을 이용해 사용합니다. 개발자는 각 agent에 맞는 Dockerfile을 이용하여, docker image를 생성하고 ECR에 업로드 합니다. 이후 bedrock-agentcore-control의 create_agent_runtime.py을 이용해서 AgentCore의 runtime으로 배포합니다. 이 작업이 끝나면 EC2와 같은 compute에 있는 streamlit에서 LangGraph와 Strands agent를 활용할 수 있습니다. 애플리케이션에서 AgentCore의 runtime을 호출할 때에는 bedrock-agentcore의 invoke_agent_runtime을 이용합니다. 이때에 각 agent를 생성할 때에 확인할 수 있는 agentRuntimeArn을 이용합니다. Agent는 MCP을 이용해 RAG, AWS Document, Tavily와 같은 검색 서비스를 활용할 수 있습니다. 여기에서는 RAG를 위하여 Lambda를 이용합니다. 데이터 저장소의 관리는 Knowledge base를 사용하고, 벡터 스토어로는 OpenSearch를 이용합니다. Agent에 필요한 S3, CloudFront, OpenSearch, Lambda등의 배포를 위해서는 AWS CDK를 이용합니다.
AgentCore의 runtime은 배포를 위해 Docker를 이용합니다. 현재(2025.7) 기준으로 arm64와 1GB 이하의 docker image를 지원합니다.
- AgentCore Runtime: AI agent와 tool을 배포하고 트래픽에 따라 자동으로 확장(Scaling)이 가능한 serverless runtime입니다. LangGraph, CrewAI, Strands Agents를 포함한 다양한 오픈소스 프레임워크을 지원합니다. 빠른 cold start, 세션 격리, 내장된 신원 확인(built-in identity), multimodal payload를 지원합니다. 이를 통해 안전하고 빠른 출시가 가능합니다.
- AgentCore Memory: Agent가 편리하게 short term, long term 메모리를 관리할 수 있습니다.
- AgentCore Code Interpreter: 분리된 sandbox 환경에서 안전하게 코드를 실행할 수 있습니다.
- AgentCore Broswer: 브라우저를 이용해 빠르고 안전하게 웹크롤링과 같은 작업을 수행할 수 있습니다.
- AgentCore Gateway: API, Lambda를 비롯한 서비스들을 쉽게 Tool로 활용할 수 있습니다.
- AgentCore Observability: 상용 환경에서 개발자가 agent의 동작을 trace, debug, monitor 할 수 있습니다.
LangGraph와 strands agent에 대한 이미지를 Dockerfile을 이용해 빌드후 ECR에 배포합니다. push-to-ecr.sh를 이용하면 손쉽게 배포할 수 있습니다.
./push-to-ecr.sh
이후, 아래와 같이 create_agent_runtime.py를 이용해 AgentCore에 runtime으로 배포합니다.
python create_agent_runtime.py
create_agent_runtime.py에서는 AgentCore에 처음으로 배포하는지 확인하여 아래와 같이 runtime을 생성합니다.
response = client.create_agent_runtime(
agentRuntimeName=runtime_name,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f"{accountId}.dkr.ecr.{aws_region}.amazonaws.com/{repositoryName}:{imageTags}"
}
},
networkConfiguration={"networkMode":"PUBLIC"},
roleArn=agent_runtime_role
)
agentRuntimeArn = response['agentRuntimeArn']기존에 runtime이 있는지는 아래와 같이 list_agent_runtimes을 이용해 확인합니다.
client = boto3.client('bedrock-agentcore-control', region_name=aws_region)
response = client.list_agent_runtimes()
isExist = False
agentRuntimeId = None
agentRuntimes = response['agentRuntimes']
targetAgentRuntime = repositoryName
if len(agentRuntimes) > 0:
for agentRuntime in agentRuntimes:
agentRuntimeName = agentRuntime['agentRuntimeName']
if agentRuntimeName == targetAgentRuntime:
agentRuntimeId = agentRuntime['agentRuntimeId']
isExist = True
break이미 runtime이 있다면 아래와 같이 update_agent_runtime을 이용해 업데이트 합니다.
response = client.update_agent_runtime(
agentRuntimeId=agentRuntimeId,
description="Update agent runtime",
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f"{accountId}.dkr.ecr.{aws_region}.amazonaws.com/{targetAgentRuntime}:{imageTags}"
}
},
roleArn=agent_runtime_role,
networkConfiguration={"networkMode":"PUBLIC"},
protocolConfiguration={"serverProtocol":"HTTP"}
)build-docker.sh와 run-docker.sh을 이용해 local 환경에서 docker 동작을 확인할 수 있습니다.
./build-docker.sh
./run-docker.sh
이후 curl.sh과 같이 동작을 테스트 할 수 있습니다.
./curl.sh
curl.sh을 이용하면 아래와 같이 local에서 테스트 할 수 있습니다. MCP server와 model 정보를 질문과 함께 제공합니다.
curl -X POST http://localhost:8080/invocations \
-H "Content-Type: application/json" \
-d '{"prompt": "내 s3 bucket 리스트는?", "mcp_servers": ["basic", "use_aws", "tavily-search", "filesystem", "terminal"], "model_name": "Claude 3.7 Sonnet"}'
invoke_agent.py와 같이 코드로도 동작으로 확인할 수 있습니다.
python invoke_agent.py
invoke_agent.py에서는 아래와 같이 invoke_agent_runtime을 이용하여 실행합니다.
payload = json.dumps({
"prompt": "서울 날씨는?",
"mcp_servers": ["basic", "use_aws", "tavily-search", "filesystem", "terminal"],
"model_name": "Claude 3.7 Sonnet",
})
agent_core_client = boto3.client('bedrock-agentcore', region_name=region_name)
response = agent_core_client.invoke_agent_runtime(
agentRuntimeArn=agentRuntimeArn,
runtimeSessionId=str(uuid.uuid4()),
payload=payload,
qualifier="DEFAULT"
)
response_body = response['response'].read()
response_data = json.loads(response_body)Streamlit에서 아래와 같이 "Docker"를 선택하면, local의 docker를 테스트 할 수 있습니다.
"Docker"를 선택하면, chat.py와 같이 http://localhost:8080/invocations 로 요청을 보내서 응답을 확인합니다.
import requests
payload = json.dumps({
"prompt": prompt, "mcp_servers": mcp_servers, "model_name": model_name,
})
headers = {"Content-Type": "application/json"}
destination = f"http://localhost:8080/invocations"
response = requests.post(destination, headers=headers, data=payload, timeout=300)문제 발생시 Docker 로그를 아래와 같이 확인합니다.
sudo docker logs coreagent-langgraph-container
Agent의 동작 테스트를 위해 S3, CloudFront, OpenSearch (Serverless), Bedrock Knowledge Base이 필요합니다. 이를 위한 상세 내용은 cdk-agentcore을 참조합니다. 이를 인프라로 배포할 때에는 아래와 같이 수행합니다.
먼저, cdk-agentcore로 이동하여 CDK 환경설정을 준비합니다. 만약 한번도 bootstrapping을 하지 않았다면, AWS CDK 부트스트래핑을 참조하여 수행합니다.
- Bootstrapping
여기서 account-id를 확인하여 아래의 "123456789012"을 바꾼후 실행합니다.
cdk bootstrap aws://123456789012/us-west-2
- CDK 배포
cd cdk-agentcore && npm install
cdk deploy --require-approval never --all
배포가 완료되면 아래와 같은 Output 파일에서 CdkAgentcoreStack.environmentforagentcore 을 복사하여 langgraph와 strands 폴더에 config.json로 업데이트 합니다.
Knowledge Base에서 문서를 활용하기 위해서는 S3에 문서 등록 및 동기화기 필요합니다. Streamlit에서 파일을 입력하면 자동으로 동기화가 시작되지만 S3로 파일을 직접 올리는 경우에는 아래와 같이 수행합니다. S3 Console에 접속하여 "storage-for-agentcore-xxxxxxxxxxxx-us-west-2"를 선택하고, 아래와 같이 docs폴더를 생성한 후에 파일을 업로드 합니다.
이후 Knowledge Bases Console에 접속하여, "agentcore"라는 Knowledge Base를 선택합니다. 이후 아래와 같이 [Sync]를 선택합니다.
AgentCore는 SSE 방식의 stream을 제공합니다.
LangGraph - agent.py와 같이 stream 방식으로 처리하면 agent가 좀 더 동적으로 동작하게 할 수 있습니다. 아래와 같이 MCP 서버의 정보로 json 파일을 만든 후에 MultiServerMCPClient으로 client를 설정하고 나서 agent를 생성합니다. 이후 stream을 이용해 출력할때 json 형태의 결과값을 stream으로 전달합니다.
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@app.entrypoint
async def agent_langgraph(payload):
mcp_json = mcp_config.load_selected_config(mcp_servers)
server_params = load_multiple_mcp_server_parameters(mcp_json)
client = MultiServerMCPClient(server_params)
app = buildChatAgentWithHistory(tools)
config = {
"recursion_limit": 50,
"configurable": {"thread_id": user_id},
"tools": tools
}
inputs = {
"messages": [HumanMessage(content=query)]
}
value = None
async for output in app.astream(inputs, config):
for key, value in output.items():
logger.info(f"--> key: {key}, value: {value}")
if "messages" in value:
for message in value["messages"]:
if isinstance(message, AIMessage):
yield({'data': message.content})
tool_calls = message.tool_calls
if tool_calls:
for tool_call in tool_calls:
tool_name = tool_call["name"]
tool_content = tool_call["args"]
toolUseId = tool_call["id"]
yield({'tool': tool_name, 'input': tool_content, 'toolUseId': toolUseId})
elif isinstance(message, ToolMessage):
toolResult = message.content
toolUseId = message.tool_call_id
yield({'toolResult': toolResult, 'toolUseId': toolUseId})Strands - agent.py와 같이 stream으로 처리합니다. 아래와 같이 AgentCore를 endpoint로 지정할 때에 agent_stream의 값을 yeild로 전달하면 streamlit 같은 client에서 동적으로 응답을 받을 수 있습니다.
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
@app.entrypoint
async def agentcore_strands(payload):
# initiate agent
await initiate_agent(
system_prompt=None,
strands_tools=strands_tools,
mcp_servers=mcp_servers,
historyMode='Disable'
)
# run agent
with mcp_manager.get_active_clients(mcp_servers) as _:
agent_stream = agent.stream_async(query)
async for event in agent_stream:
text = ""
if "data" in event:
text = event["data"]
stream = {'data': text}
elif "result" in event:
final = event["result"]
message = final.message
if message:
content = message.get("content", [])
result = content[0].get("text", "")
stream = {'result': result}
elif "current_tool_use" in event:
current_tool_use = event["current_tool_use"]
name = current_tool_use.get("name", "")
input = current_tool_use.get("input", "")
toolUseId = current_tool_use.get("toolUseId", "")
text = f"name: {name}, input: {input}"
stream = {'tool': name, 'input': input, 'toolUseId': toolUseId}
elif "message" in event:
message = event["message"]
if "content" in message:
content = message["content"]
if "toolResult" in content[0]:
toolResult = content[0]["toolResult"]
toolUseId = toolResult["toolUseId"]
toolContent = toolResult["content"]
toolResult = toolContent[0].get("text", "")
stream = {'toolResult': toolResult, 'toolUseId': toolUseId}
yield (stream)AgentCore로 agent_runtime_arn을 이용해 request에 대한 응답을 얻습니다. 이때 content-type이 "text/event-stream"인 경우에 prefix인 "data:"를 제거한 후에 json parser를 이용해 얻어진 값을 목적에 맞게 활용합니다.
agent_core_client = boto3.client('bedrock-agentcore', region_name=bedrock_region)
response = agent_core_client.invoke_agent_runtime(
agentRuntimeArn=agent_runtime_arn,
runtimeSessionId=runtime_session_id,
payload=payload,
qualifier="DEFAULT" # DEFAULT or LATEST
)
result = current = ""
processed_data = set() # Prevent duplicate data
# stream response
if "text/event-stream" in response.get("contentType", ""):
for line in response["response"].iter_lines(chunk_size=10):
line = line.decode("utf-8")
if line.startswith('data: '):
data = line[6:].strip() # Remove "data:" prefix and whitespace
if data: # Only process non-empty data
# Check for duplicate data
if data in processed_data:
continue
processed_data.add(data)
data_json = json.loads(data)
if 'data' in data_json:
text = data_json['data']
logger.info(f"[data] {text}")
current += text
containers['result'].markdown(current)
elif 'result' in data_json:
result = data_json['result']
elif 'tool' in data_json:
tool = data_json['tool']
input = data_json['input']
toolUseId = data_json['toolUseId']
if toolUseId not in tool_info_list: # new tool info
tool_info_list[toolUseId] = index
add_notification(containers, f"Tool: {tool}, Input: {input}")
else: # overwrite tool info
containers['notification'][tool_info_list[toolUseId]].info(f"Tool: {tool}, Input: {input}")
elif 'toolResult' in data_json:
toolResult = data_json['toolResult']
toolUseId = data_json['toolUseId']
if toolUseId not in tool_result_list: # new tool result
tool_result_list[toolUseId] = index
add_notification(containers, f"Tool Result: {toolResult}")
else: # overwrite tool result
containers['notification'][tool_result_list[toolUseId]].info(f"Tool Result: {toolResult}")여기서는 Streamlit을 이용하여 AgentCore의 동작을 테스트 할 수 있습니다. 아래와 streamlit을 실행할 수 있습니다.
streamlit run application/app.py
실행 후에 아래와 같이 왼쪽 메뉴에서 사용할 MCP 서버를 선택하고 질문을 입력합니다.
MCP server에서 "use_aws"를 선택하고, "내 cloudwatch 로그 리스트는?"라고 입력하면 AWS CLI를 이용해 AWS cloudwatch의 로그 리스트를 확인하여 아래와 같이 보여줍니다.
"tavily search"를 선택하고, "강남역 맛집은?"이라고 검색하면 아래와 같이 강남역에 대한 정보를 검색하여 얻어진 결과를 보여줍니다.
Get started with the Amazon Bedrock AgentCore Runtime starter toolkit
Amazon Bedrock AgentCore - Developer Guide
BedrockAgentCoreControlPlaneFrontingLayer
Amazon Bedrock AgentCore Samples
Amazon Bedrock AgentCore RuntCode Interpreter
Add observability to your Amazon Bedrock AgentCore resources
Hosting Strands Agents with Amazon Bedrock models in Amazon Bedrock AgentCore Runtime