在 AgentCore 執行期中部署 AG-UI 伺服器

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

建立package.json第一個：

{
  "name": "my-agui-server",
  "type": "module",
  "scripts": {
    "build": "tsc"
  },
  "dependencies": {
    "@ag-ui/aws-strands": "^0.1.0",
    "@strands-agents/sdk": "^1.1.0"
  },
  "devDependencies": {
    "@types/express": "^5.0.0",
    "@types/node": "^22.0.0",
    "tsx": "^4.0.0",
    "typescript": "^5.0.0"
  }
}

然後安裝相依性：

npm install

建立名為 my_agui_server.py 的新檔案。此範例使用 AWS Strands 搭配 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)

建立名為 my-agui-server.ts 的新檔案。此範例使用 AWS Strands 搭配 AG-UI：

// my-agui-server.ts
import { Agent } from "@strands-agents/sdk";
import { StrandsAgent } from "@ag-ui/aws-strands";
import { createStrandsApp } from "@ag-ui/aws-strands/server";

async function main(): Promise<void> {
  // Create a simple Strands agent
  const strandsAgent = new Agent({
    systemPrompt: "You are a helpful assistant.",
  });

  // Wrap with AG-UI protocol support
  const aguiAgent = new StrandsAgent({
    agent: strandsAgent,
    name: "my_agent",
    description: "A helpful assistant",
  });

  // Express app exposing the AgentCore-required paths on port 8080
  const app = await createStrandsApp(aguiAgent, {
    path: "/invocations",
    pingPath: "/ping",
  });

  app.listen(8080, () => {
    console.log("AG-UI server running on port 8080");
  });
}

void main();

python my_agui_server.py

npx tsx my-agui-server.ts

## 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

## Project Folder Structure
your_project_directory/
├── my-agui-server.ts          # Your main agent code
├── package.json               # Dependencies for your agent
└── tsconfig.json              # TypeScript compiler configuration

建立 tsconfig.json ：

{
  "compilerOptions": {
    "target": "ES2022",
    "lib": ["ES2022", "DOM"],
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "outDir": "./dist",
    "strict": true,
    "esModuleInterop": true
  },
  "include": ["*.ts"]
}

agentcore configure -e my_agui_server.py --protocol AGUI

agentcore configure -e my-agui-server.ts --protocol AGUI

安裝必要的套件：

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!"))

安裝必要的套件：

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!");