Amazon Bedrock AgentCore가 AG-UI를 지원하는 방법 AgentCore 런타임에서 AG-UI 사용 부록

AgentCore 런타임에 AG-UI 서버 배포

Amazon Bedrock AgentCore 런타임을 사용하면 AgentCore 런타임에서 에이전트 사용자 인터페이스(AG-UI) 서버를 배포하고 실행할 수 있습니다. 이 가이드에서는 첫 번째 AG-UI 서버를 생성, 테스트 및 배포하는 방법을 안내합니다.

이 섹션에서는 다음을 배웁니다.

Amazon Bedrock AgentCore가 AG-UI를 지원하는 방법
AG-UI 서버를 생성하는 방법
로컬에서 서버를 테스트하는 방법
에 서버를 배포하는 방법 AWS
배포된 서버를 호출하는 방법

AG-UI에 대한 자세한 내용은 AG-UI 프로토콜 계약을 참조하세요.

Amazon Bedrock AgentCore가 AG-UI를 지원하는 방법

Amazon Bedrock AgentCore의 AG-UI 프로토콜 지원을 통해 프록시 계층 역할을 하여 에이전트 사용자 인터페이스 서버와 통합할 수 있습니다. AG-UI용으로 구성된 경우 Amazon Bedrock AgentCore는 컨테이너가 HTTP/SSE 또는 WebSocket 연결 /invocations 경로8080의 포트에서 서버를 실행할 것으로 예상/ws합니다. AG-UI는 HTTP 프로토콜과 동일한 포트와 경로를 사용하지만 런타임은 배포 구성 중에 지정된 --protocol 플래그를 기반으로 포트와 경로를 구분합니다.

Amazon Bedrock AgentCore는 클라이언트와 AG-UI 컨테이너 간의 프록시 역할을 합니다. InvokeAgentRuntime API의 요청은 수정 없이 컨테이너로 전달됩니다. Amazon Bedrock AgentCore는 인증(SigV4/OAuth 2.0), 세션 격리 및 조정을 처리합니다.

다른 프로토콜과의 주요 차이점:

포트: AG-UI 서버는 포트 8080에서 실행됩니다(HTTP와 동일, MCP의 경우 8000, A2A의 경우 9000).
경로: AG-UI 서버는 /invocations HTTP/SSE 및 WebSocket/ws에를 사용합니다(HTTP 프로토콜과 동일).
메시지 형식: 스트리밍을 위해 Server-Sent Events(SSE)를 통해 이벤트 스트림을 사용하거나 양방향 통신을 위해 WebSocket을 사용합니다.
프로토콜 포커스: Agent-to-User 상호 작용(도구의 경우 MCP, agent-to-agent의 경우 A2A)
Authentication: SigV4 및 OAuth 2.0 인증 체계 모두 지원

자세한 내용은 https://docs.ag-ui.com/introduction 단원을 참조하십시오.

AgentCore 런타임에서 AG-UI 사용

이 자습서에서는 AG-UI 서버를 생성, 테스트 및 배포합니다.

전체 예제 및 프레임워크별 구현은 AG-UI Quickstart 설명서 및 AG-UI Dojo를 참조하세요.

주제

사전 조건
1단계: AG-UI 서버 생성
2단계: 로컬에서 AG-UI 서버 테스트
3단계: Bedrock AgentCore 런타임에 AG-UI 서버 배포
4단계: 배포된 AG-UI 서버 호출

사전 조건

Python 3.12 이상 설치됨
AgentCore CLI에 설치된 Node.js 20 이상
적절한 권한과 로컬 자격 증명이 구성된 AWS 계정
AG-UI 프로토콜 및 이벤트 기반 agent-to-user 통신 개념 이해

1단계: AG-UI 서버 생성

AG-UI는 여러 에이전트 프레임워크에서 지원됩니다. 이 자습서에서는 Python용 AWS Strands를 사용합니다.

필수 패키지 설치

AG-UI를 지원하는 AWS Strands용 패키지를 설치합니다.


pip install fastapi
pip install uvicorn
pip install ag-ui-strands

다른 프레임워크는 AG-UI 프레임워크 통합을 참조하세요.

첫 번째 AG-UI 서버 생성

my_agui_server.py이라는 이름의 파일을 만듭니다. 이 예제에서는 AG-UI와 함께 AWS Strands를 사용합니다. 서버는 포트에서 수신 대기하고 8080AG-UI 트래픽을 /invocations 노출하며 상태 확인을 /ping 위해를 노출합니다. AgentCore 런타임에는 AG-UI 컨테이너에 대해이 계약이 필요합니다.


# my_agui_server.py
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from ag_ui_strands import StrandsAgent
from ag_ui.core import RunAgentInput
from ag_ui.encoder import EventEncoder
from strands import Agent

# Create a simple Strands agent
strands_agent = Agent(
    system_prompt="You are a helpful assistant.",
)

# Wrap with AG-UI protocol support
agui_agent = StrandsAgent(
    agent=strands_agent,
    name="my_agent",
    description="A helpful assistant",
)

# FastAPI server
app = FastAPI()

@app.post("/invocations")
async def invocations(input_data: dict, request: Request):
    """Main AG-UI endpoint that returns event streams."""
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        run_input = RunAgentInput(**input_data)
        async for event in agui_agent.run(run_input):
            yield encoder.encode(event)

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

@app.get("/ping")
async def ping():
    return JSONResponse({"status": "Healthy"})

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

프레임워크별 전체 예제는 다음을 참조하세요.

코드 이해

이벤트 스트림: AG-UI는 Server-Sent Events(SSE)를 사용하여 형식이 지정된 이벤트를 클라이언트로 스트리밍합니다.
/invocations 엔드포인트: HTTP/SSE 통신을 위한 기본 엔드포인트(HTTP 프로토콜과 동일)
포트 8080: AG-UI 서버는 AgentCore 런타임에서 기본적으로 포트 8080에서 실행됩니다.

2단계: 로컬에서 AG-UI 서버 테스트

로컬 개발 환경에서 AG-UI 서버를 실행하고 테스트합니다.

AG-UI 서버 시작

AG-UI 서버를 로컬에서 실행합니다.


python my_agui_server.py

서버가 포트에서 실행 중임을 나타내는 출력이 표시되어야 합니다8080.

엔드포인트 테스트

올바른 형식의 AG-UI 요청으로 SSE 엔드포인트를 테스트합니다.


curl -N -X POST http://localhost:8080/invocations \
-H "Content-Type: application/json" \
-d '{
  "threadId": "test-123",
  "runId": "run-456",
  "state": {},
  "messages": [{"role": "user", "content": "Hello, agent!", "id": "msg-1"}],
  "tools": [],
  "context": [],
  "forwardedProps": {}
}'

, 및 이벤트를 포함하여 SSE 형식으로 반환되는 AG-UI RUN_FINISHED 이벤트 스트림RUN_STARTEDTEXT_MESSAGE_CONTENT이 표시되어야 합니다.

3단계: Bedrock AgentCore 런타임에 AG-UI 서버 배포

AgentCore CLI를 AWS 사용하여에 AG-UI 서버를 배포합니다.

배포 도구 설치

AgentCore CLI를 설치합니다.


npm install -g @aws/agentcore

먼저 다음 구조의 프로젝트 폴더를 생성합니다.


## Project Folder Structure
your_project_directory/
├── my_agui_server.py          # Your main agent code
├── requirements.txt           # Dependencies for your agent

종속성을 requirements.txt 사용하여 라는 새 파일을 생성합니다.


fastapi
uvicorn
ag-ui-strands

인증을 위한 Cognito 사용자 풀 설정

배포된 서버에 대한 보안 액세스를 위한 인증을 구성합니다. 자세한 Cognito 설정 지침은 인증을 위한 Cognito 사용자 풀 설정을 참조하세요. 이렇게 하면 배포된 서버에 대한 보안 액세스에 필요한 OAuth 토큰이 제공됩니다.

Cognito 설정을 완료한 후 배포 명령에서 사용하는 값을 내보냅니다.


export REGION="<your-region>"
export POOL_ID="<your-user-pool-id>"
export CLIENT_ID="<your-app-client-id>"

배포를 위해 AG-UI 서버 구성

빈 AgentCore 프로젝트를 생성합니다. 그런 다음 첫 번째 AG-UI 서버 생성에서 생성한 서버를 이전 단계의 Cognito 구성을 사용하여 BYO 에이전트로 등록합니다.


agentcore create --project-name AguiProject --no-agent
cd AguiProject
agentcore add agent \
  --name AguiAgent \
  --type byo \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --code-location .. \
  --entrypoint my_agui_server.py \
  --protocol AGUI \
  --authorizer-type CUSTOM_JWT \
  --discovery-url "https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" \
  --allowed-clients "$CLIENT_ID" \
  --request-header-allowlist Authorization

명령은 기존 구현을 이전 단계의 AG-UI 프로토콜 및 Cognito OAuth 구성에 등록합니다.

에 배포 AWS

에이전트를 배포합니다.


agentcore deploy

배포 후 다음과 같은 에이전트 런타임 ARN을 받게 됩니다.


arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_agui_server-xyz123

4단계: 배포된 AG-UI 서버 호출

배포된 Amazon Bedrock AgentCore AG-UI 서버를 호출하고 이벤트 스트림과 상호 작용합니다.

환경 변수 설정

보유자 토큰을 환경 변수로 내보냅니다. 보유자 토큰 설정은 인증을 위한 Cognito 사용자 풀 설정을 참조하세요.
```
export BEARER_TOKEN="<BEARER_TOKEN>"
```

에이전트 ARN을 내보냅니다.


export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_agui_server-xyz123"

AG-UI 서버 호출

AG-UI 서버를 프로그래밍 방식으로 호출하려면 클라이언트와 일치하는 언어를 선택합니다.

예

Python

필수 패키지를 설치합니다.


pip install httpx httpx-sse

그런 다음 다음 클라이언트 코드를 사용합니다.


import asyncio
import json
import os
from urllib.parse import quote
from uuid import uuid4

import httpx
from httpx_sse import aconnect_sse

async def invoke_agui_agent(message: str):
    agent_arn = os.environ.get('AGENT_ARN')
    bearer_token = os.environ.get('BEARER_TOKEN')
    escaped_arn = quote(agent_arn, safe='')

    url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_arn}/invocations?qualifier=DEFAULT"
    headers = {
        "Authorization": f"Bearer {bearer_token}",
        "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": str(uuid4()),
    }
    payload = {
        "threadId": str(uuid4()),
        "runId": str(uuid4()),
        "messages": [{"id": str(uuid4()), "role": "user", "content": message}],
        "state": {},
        "tools": [],
        "context": [],
        "forwardedProps": {},
    }

    async with httpx.AsyncClient(timeout=300) as client:
        async with aconnect_sse(client, "POST", url, headers=headers, json=payload) as sse:
            async for event in sse.aiter_sse():
                data = json.loads(event.data)
                event_type = data.get("type")
                if event_type == "TEXT_MESSAGE_CONTENT":
                    print(data.get("delta", ""), end="", flush=True)
                elif event_type == "RUN_ERROR":
                    print(f"Error: {data.get('code')} - {data.get('message')}")

asyncio.run(invoke_agui_agent("Hello!"))

TypeScript

필수 패키지를 설치합니다.


npm install @ag-ui/client

그런 다음 다음 클라이언트 코드를 사용합니다.


import { HttpAgent, AgentSubscriber } from "@ag-ui/client";
import { randomUUID } from "crypto";

async function invokeAguiAgent(message: string): Promise<void> {
  const agentArn = process.env.AGENT_ARN!;
  const bearerToken = process.env.BEARER_TOKEN!;
  const escapedArn = encodeURIComponent(agentArn);

  const agent = new HttpAgent({
    url: `https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${escapedArn}/invocations?qualifier=DEFAULT`,
    headers: {
      Authorization: `Bearer ${bearerToken}`,
      "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": randomUUID(),
    },
  });

  agent.messages = [{ id: randomUUID(), role: "user", content: message }];

  const subscriber: AgentSubscriber = {
    onTextMessageContentEvent: ({ event }) => {
      process.stdout.write(event.delta);
    },
    onRunErrorEvent: ({ event }) => {
      console.error(`Error: ${event.code ?? "RUN_ERROR"} - ${event.message}`);
    },
  };

  await agent.runAgent({}, subscriber);
}

void invokeAguiAgent("Hello!");