AgentCore CLI で自作したAIエージェントを GenU に統合してみた

agentcore create   # プロジェクト作成
agentcore dev      # ローカル開発サーバー起動
agentcore deploy   # AWS へデプロイ
agentcore invoke   # デプロイ済みエージェントの実行

npm install -g @aws/agentcore

agentcore create

Name:       WeatherAgent
Language:   Python
Build:      Container
Protocol:   HTTP
Framework:  Strands
Model:      Bedrock (us.anthropic.claude-sonnet-4-5-20250514-v1:0)
Memory:     None
Network:    PUBLIC

cd WeatherAgent

WeatherAgent/
├── agentcore/
│   ├── agentcore.json      # リソース設定
│   ├── aws-targets.json    # デプロイ先設定
│   └── cdk/                # CDK インフラ（自動生成）
└── app/
    └── MyAgent/
        └── main.py         # エージェントのコード

from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model
from mcp_client.client import get_streamable_http_mcp_client
from memory.session import get_memory_session_manager

app = BedrockAgentCoreApp()
log = app.logger

# MCP クライアントの定義
mcp_clients = [get_streamable_http_mcp_client()]

# ツールの定義
tools = []

@tool
def add_numbers(a: int, b: int) -> int:
    """Return the sum of two numbers"""
    return a+b
tools.append(add_numbers)

for mcp_client in mcp_clients:
    if mcp_client:
        tools.append(mcp_client)

def agent_factory():
    cache = {}
    def get_or_create_agent(session_id, user_id):
        key = f"{session_id}/{user_id}"
        if key not in cache:
            cache[key] = Agent(
                model=load_model(),
                session_manager=get_memory_session_manager(session_id, user_id),
                system_prompt="You are a helpful assistant. Use tools when appropriate.",
                tools=tools
            )
        return cache[key]
    return get_or_create_agent
get_or_create_agent = agent_factory()

@app.entrypoint
async def invoke(payload, context):
    log.info("Invoking Agent.....")
    session_id = getattr(context, 'session_id', 'default-session')
    user_id = getattr(context, 'user_id', 'default-user')
    agent = get_or_create_agent(session_id, user_id)

    stream = agent.stream_async(payload.get("prompt"))
    async for event in stream:
        if "data" in event and isinstance(event["data"], str):
            yield event["data"]

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

import json
import urllib.request

from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model

app = BedrockAgentCoreApp()
log = app.logger

@tool
def get_weather(city: str) -> str:
    """指定された都市の現在の天気を取得します。

    Args:
        city: 天気を取得したい都市名（例: Tokyo, New York, London）
    """
    # 都市名から緯度経度を取得（Open-Meteo Geocoding API）
    geo_url = f"https://geocoding-api.open-meteo.com/v1/search?name={city}&count=1&language=ja"
    with urllib.request.urlopen(geo_url) as response:
        geo_data = json.loads(response.read())

    if "results" not in geo_data:
        return f"都市 '{city}' が見つかりませんでした。"

    location = geo_data["results"][0]
    lat = location["latitude"]
    lon = location["longitude"]
    name = location.get("name", city)

    # 天気情報を取得（Open-Meteo Weather API）
    weather_url = (
        f"https://api.open-meteo.com/v1/forecast?"
        f"latitude={lat}&longitude={lon}"
        f"&current=temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m"
        f"&timezone=Asia/Tokyo"
    )
    with urllib.request.urlopen(weather_url) as response:
        weather_data = json.loads(response.read())

    current = weather_data["current"]
    return (
        f"{name} の現在の天気\n"
        f"気温: {current['temperature_2m']}°C\n"
        f"湿度: {current['relative_humidity_2m']}%\n"
        f"風速: {current['wind_speed_10m']} km/h"
    )

def process_prompt(prompt):
    """GenU が送る prompt（ContentBlock 配列）からテキストを抽出する。"""
    if isinstance(prompt, str):
        return prompt
    if isinstance(prompt, list):
        texts = [block["text"] for block in prompt if isinstance(block, dict) and "text" in block]
        return "\n".join(texts) if texts else ""
    return str(prompt)

@app.entrypoint
async def invoke(payload):
    log.info("Invoking Agent.....")

    # GenUのフロントエンドが送るペイロードから会話履歴と現在の入力を取得
    messages = payload.get("messages", [])
    prompt = process_prompt(payload.get("prompt", ""))

    agent = Agent(
        model=load_model(),
        system_prompt="あなたは天気予報アシスタントです。ユーザーが都市名を伝えると、現在の天気を取得して日本語で回答します。",
        tools=[get_weather],
        messages=messages,
    )
    # GenUのフロントエンドにレスポンス形式を合わせる
    async for event in agent.stream_async(prompt):
        if "event" in event:
            yield event

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

agentcore dev

agentcore deploy

agentcore status

arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/xxxxxxxx

agentcore invoke "今日の大阪の天気を教えてください" --agent MyAgent

const envs: Record<string, Partial<StackInput>> = {
  myenv: {
    modelRegion: "us-east-1",
    createGenericAgentCoreRuntime: true,
    agentBuilderEnabled: true,
    agentCoreExternalRuntimes: [
      {
        name: 'WeatherAgent',
        arn: 'arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/xxxxxxxx',
        description: 'This is WeatherAgent.',
      },
    ],
  },
};

npm run cdk:deploy