여기에서는 AgentCore Runtime을 이용해서 1) LangGraph, Strands SDK, Claude Agent SDK를 이용해 만든 agent를 배포하는 방법과 2) agent에 필요한 데이터를 수집하고 사용자의 의도에 따른 동작을 수행하는 방법을 설명합니다. AgentCore는 Agent와 MCP를 위한 서버리스 production 환경으로서 Agent와 MCP 서버를 편리하게 배포하고 안전하고 효과적으로 운용할 수 있습니다.
전체적인 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
push-to-ecr.sh에서는 아래와 같이 ECR에 login후에 push를 수행합니다.
aws ecr get-login-password --region ${AWS_REGION} | docker login --username AWS --password-stdin ${AWS_ACCOUNT_ID}.dkr.ecr.${AWS_REGION}.amazonaws.com
docker build -t ${ECR_REPOSITORY}:${IMAGE_TAG} .
docker tag ${ECR_REPOSITORY}:${IMAGE_TAG} ${ECR_URI}
docker push ${ECR_URI}
이후, 아래와 같이 create_agent_runtime.py를 이용해 AgentCore에 runtime으로 배포합니다.
python create_agent_runtime.py
create_agent_runtime.py에서는 AgentCore에 처음으로 배포하는지 확인하여 아래와 같이 runtime을 생성합니다. 여기서 networkMode는 PUBLIC/VPC를 선택할 수 있어서 필요시 agent를 특정 VPC 접속으로 제한할 수 있고, Security Group을 이용하여 사내로 접속을 제한할 수 있습니다. 또한, protocolConfiguration은 HTTP, MCP, A2A를 선택하여 필요한 용도에 맞게 사용할 수 있습니다. 인증은 기본이 IAM이며, 필요시 authorizerConfiguration을 이용해 JWT를 사용할 수 있습니다.
response = client.create_agent_runtime(
agentRuntimeName=runtime_name,
agentRuntimeArtifact={
'containerConfiguration': {
'containerUri': f"{accountId}.dkr.ecr.{aws_region}.amazonaws.com/{repositoryName}:{imageTags}"
}
},
networkConfiguration={"networkMode":"PUBLIC"},
protocolConfiguration={"serverProtocol":"HTTP"}
roleArn=agent_runtime_role
)
agentRuntimeArn = response['agentRuntimeArn']Runtime agent를 생성하기 전에 기존 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"}
)Agent를 AgentCore에 배포하기 전에 local 환경에서 충분히 동작을 테스트하면 개발 소요시간을 단축할 수 있습니다. build-docker.sh를 이용해 local 환경에서 docker로 된 runtime을 빌드합니다.
./build-docker.sh
build-docker.sh에서는 config.json에서 환경 값을 읽은 후에 아래와 같이 docker를 빌드합니다. docker에서 AWS CLI를 사용하기 위해 AWS Credential을 환경값으로 주고 있습니다.
sudo docker build \
--platform linux/arm64 \
--build-arg AWS_ACCESS_KEY_ID="$AWS_ACCESS_KEY_ID" \
--build-arg AWS_SECRET_ACCESS_KEY="$AWS_SECRET_ACCESS_KEY" \
--build-arg AWS_DEFAULT_REGION="${AWS_DEFAULT_REGION:-us-east-1}" \
--build-arg AWS_SESSION_TOKEN="$AWS_SESSION_TOKEN" \
-t ${DOCKER_NAME}:latest .
run-docker.sh을 이용해 실행할 수 있습니다.
./run-docker.sh
Runtime agent의 경우에 run-docker.sh와 같이 8080으로 expose를 합니다.
docker run -d \
--name ${DOCKER_NAME}-container \
-p 8080:8080 \
--entrypoint="" \
${DOCKER_NAME}:latest \
uv run uvicorn agent:app --host 0.0.0.0 --port 8080
이후 test_runtime_local.py을 이용해 동작을 테스트 합니다.
python test_runtime_local.py
test_runtime_local.py에서는 아래와 같이 prompt, MCP server, model 정보를 설정합니다. 이후 request를 POST로 전송한 후에 결과를 확인합니다.
prompt = "보일러 에러 코드?"
mcp_servers = ["kb-retriever"]
model_name = "Claude 3.7 Sonnet"
user_id = uuid.uuid4().hex
history_mode = "Disable"
payload = {
"prompt": prompt,
"mcp_servers": mcp_servers,
"model_name": model_name,
"user_id": user_id,
"history_mode": history_mode
}
runtime_url = "http://127.0.0.1:8080/invocations"
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
response = requests.post(runtime_url, headers=headers, json=payload, stream=True)
if "text/event-stream" in response.headers.get("content-type", ""):
for line in response.iter_lines(chunk_size=10):
if line:
line = line.decode("utf-8")
print(f"-> {line}")또한, streamlit에서 아래와 같이 "Docker"를 선택하면, local의 docker를 테스트 할 수 있습니다.
"Docker"를 선택하면, agentcore_client.py와 같이 http://localhost:8080/invocations 로 요청을 보내서 응답을 확인합니다.
import requests
payload = json.dumps({
"prompt": prompt,
"mcp_servers": mcp_servers,
"model_name": model_name,
"user_id": user_id,
"history_mode": history_mode
})
headers = {"Content-Type": "application/json"}
destination = f"http://localhost:8080/invocations"
response = requests.post(destination, headers=headers, data=payload, timeout=300)문제 발생시 Docker 로그를 아래와 같이 확인합니다.
docker logs agentcore_langgraph-container -f
mcp_server_use_aws.py에서는 아래와 같이 use_aws tool을 등록합니다. use_aws tool은 agent가 전달하는 service_name, operation_name, parameters를 받아서 실행하고 결과를 리턴합니다. service_name은 s3, ec2와 같은 서비스 명이며, operation_name은 list_buckets와 같은 AWS CLI 명령어 입니다. 또한, parameters는 이 명령어를 수행하는데 필요한 값입니다.
import use_aws as aws_utils
@mcp.tool()
def use_aws(service_name, operation_name, parameters, region, label, profile_name) -> Dict[str, Any]:
console = aws_utils.create()
available_operations = get_available_operations(service_name)
client = get_boto3_client(service_name, region, profile_name)
operation_method = getattr(client, operation_name)
response = operation_method(**parameters)
for key, value in response.items():
if isinstance(value, StreamingBody):
content = value.read()
try:
response[key] = json.loads(content.decode("utf-8"))
except json.JSONDecodeError:
response[key] = content.decode("utf-8")
return {
"status": "success",
"content": [{"text": f"Success: {str(response)}"}],
}use-aws은 use_aws.py의 MCP 버전입니다.
kb-retriever를 이용해 완전관리형 RAG 서비스인 Knowledge base의 정보를 조회할 수 있습니다. mcp_server_retrieve.py에서는 agent가 전달하는 keyword를 이용해 mcp_retrieve의 retrieve를 호출합니다.
@mcp.tool()
def retrieve(keyword: str) -> str:
return mcp_retrieve.retrieve(keyword) kb-retriever는 아래와 같이 bedrock-agent-runtime를 이용하여 Knowledge Base를 조회합니다. 이때, number_of_results의 결과를 얻은 후에 content와 reference 정보를 추출하여 활용합니다.
bedrock_agent_runtime_client = boto3.client("bedrock-agent-runtime", region_name=bedrock_region)
response = bedrock_agent_runtime_client.retrieve(
retrievalQuery={"text": query},
knowledgeBaseId=knowledge_base_id,
retrievalConfiguration={
"vectorSearchConfiguration": {"numberOfResults": number_of_results},
},
)
retrieval_results = response.get("retrievalResults", [])
json_docs = []
for result in retrieval_results:
text = url = name = None
if "content" in result:
content = result["content"]
if "text" in content:
text = content["text"]
if "location" in result:
location = result["location"]
if "s3Location" in location:
uri = location["s3Location"]["uri"] if location["s3Location"]["uri"] is not None else ""
name = uri.split("/")[-1]
url = uri # TODO: add path and doc_prefix
elif "webLocation" in location:
url = location["webLocation"]["url"] if location["webLocation"]["url"] is not None else ""
name = "WEB"
json_docs.append({
"contents": text,
"reference": {
"url": url,
"title": name,
"from": "RAG"
}
})kb-retrieve MCP server는 RAG를 위해 Knowledge Base를 활용합니다. 여기서는 비용면에서 유용한 S3 Vector를 이용해 Knowledge base의 데이터 소스를 생성합니다.
S3 Vector를 생성할 때에는 s3vector.py와 같이 embedding 모델을 지정하고 S3 bucket을 이용해 동기화하도록 설정합니다.
response = bedrock_agent.create_knowledge_base(
name=knowledge_base_name,
description=f"Knowledge base for {projectName} using s3 vector",
roleArn=role_arn,
knowledgeBaseConfiguration={
"type": "VECTOR",
"vectorKnowledgeBaseConfiguration": {
"embeddingModelArn": embeddingModelArn,
"embeddingModelConfiguration": {
"bedrockEmbeddingModelConfiguration": {
"dimensions": 1024,
"embeddingDataType": "FLOAT32"
}
},
"supplementalDataStorageConfiguration": {
"storageLocations": [
{
"s3Location": {
"uri": f"s3://{bucket_name}"
},
"type": "S3"
}
]
}
}
},
storageConfiguration={
"type": "S3_VECTORS",
"s3VectorsConfiguration": {
"vectorBucketArn": s3_vector_bucket_arn,
"indexArn": s3_vector_index_arn
}
}
)이를 배포할 때에는 아래와 같이 수행합니다.
- create_iam_policies.py를 이용해 필요한 권한을 생성합니다.
python create_iam_policies.py
- MCP 접속에 활용할 congnito를 설정하고 생성된 token을 secret manager에 등록합니다.
python create_bearer_token.py
- Docker image를 생성하여 ECR에 푸쉬합니다.
./build-docker.sh
- AgentCore에 MCP server를 생성합니다.
python create_mcp_runtime.py
만약 local에서 배포 결과를 확인하고 싶다면, test_mcp_remote.py를 이용해 테스트를 수행합니다.
python test_mcp_remote.pytest_mcp_remote.py에서는 아래와 같이 mcp_url에 요청을 보내서 동작을 확인합니다. 아래에서는 kb-retrieve MCP server의 retrieve tool에 대한 동작 테스트를 수행합니다.
mcp_url = f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
headers = {
"Authorization": f"Bearer {bearer_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
async with streamablehttp_client(mcp_url, headers, timeout=120, terminate_on_close=False) as (
read_stream, write_stream, _):
params = {"keyword": "보일러 에러 코드"}
async with ClientSession(read_stream, write_stream) as session:
result = await asyncio.wait_for(session.call_tool("retrieve", params), timeout=30)
print(f"retrieve result: {result}")Knowledge Base에서 문서를 활용하기 위해서는 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
AWS re:Invent 2025 - Architecting scalable and secure agentic AI with Bedrock AgentCore (AIM431)