33
16

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

今一番アツい、AWSのAIエージェント構築サービスを触ってみましょう💪

事前準備

  • AWSアカウント作成
    • 「6ヶ月無料プラン」ではなく「有料プラン」で(費用は数十円レベル)
    • フリーメールなど捨てアドを使えば、終了後に即閉鎖できるのでオススメ

今回使うサービスは、サーバーレス中心なので維持費用はほぼかかりません!
(エージェントとお喋りするたびに、Claude APIの重量課金が数円ぐらいずつ発生するのみ)

0. セットアップ

  • AWSにログイン(東京リージョン)
  • BedrockのプレイグラウンドでSonnet 4.6を選択し、ユースケースを申請
  • CloudShellを起動
  • Pythonをバージョンアップ
sudo yum -y install python3.13 python3.13-pip
  • エイリアスを設定
alias python=python3.13
alias pip=pip3.13

1. AIエージェント入門

  • CloudShellを起動
  • Pythonをバージョンアップ
sudo yum -y install python3.13 python3.13-pip
  • エイリアスを設定
alias python=python3.13
alias pip=pip3.13

スクリーンショット 2025-10-04 18.27.44.png

  • ファイル作成
nano 1.py
  • 以下をコピペ
1.py
# 必要なライブラリをインポート
from strands import Agent

# エージェントを作成して起動
agent = Agent()
agent("KAGってどんな会社?")
  • Ctrl + x > y > Enter で閉じる
  • 実行
pip install --user strands-agents
python 1.py

2. ツール

  • Tavily APIキーを取得する(無料)

  • ファイル作成
nano 2.py
  • 以下の tvly-xxxxxxxxx 部分に自分のAPIキーを記載して、コピペ
2.py
# 必要なライブラリをインポート
from strands import Agent, tool
from tavily import TavilyClient

# Web検索関数をツールとして定義
@tool
def search(query):
    tavily = TavilyClient("tvly-xxxxxxxxxxxxxxxxxxxxxxxxxx")
    return tavily.search(query)

# ツールを設定したエージェントを作成
agent = Agent(
    model="jp.anthropic.claude-sonnet-4-6",
    tools=[search],
)

# エージェントを起動
agent("KAGってどんな会社?")

APIキーをコードにベタ書きするのは本来好ましくありません。あくまでTavilyの無料アカウントで、かつハンズオンの一時的な利用に留めましょう。

pip install --user tavily-python
python 2.py

3. ツール(MCP活用)

スクリーンショット 2025-10-04 19.48.33.png

nano 3.py
3.py
# 必要なライブラリをインポート
from strands import Agent
from strands.tools.mcp import MCPClient
from mcp.client.streamable_http import streamablehttp_client

# MCPクライアントを作成
mcp_client = MCPClient(
    lambda: streamablehttp_client(
        "https://knowledge-mcp.global.api.aws"
    )
)

# エージェントを作成
agent = Agent(
    model="jp.anthropic.claude-sonnet-4-6",
    system_prompt="AWS MCPを使って質問に答えてね",
    tools=[mcp_client]
)

# エージェントを起動
agent("StrandsとAgentCoreの役割の違いを初心者向けに短く説明して")
pip install --user strands-agents-tools
python 3.py

4. マルチエージェント

スクリーンショット 2025-10-04 20.22.36.png

nano 4.py
# 必要なライブラリをインポート
from strands import Agent, tool
from strands_tools import calculator

# サブエージェント1を定義
@tool
def math_agent(query):
    agent = Agent(
        system_prompt="ツールを使って計算を行ってください",
        tools=[calculator]
    )
    return agent(query)

# サブエージェント2を定義
@tool
def haiku_agent(query):
    agent = Agent(
        system_prompt="与えられたお題で五・七・五の俳句を詠んで"
    )
    return agent(query)

# 監督者エージェントの作成と実行
orchestrator = Agent(
    system_prompt="与えられた問題を計算して、答えを俳句として詠んで",
    tools=[math_agent, haiku_agent]
)
orchestrator("十円持っている太郎くんが二十円もらいました。今いくら?")
pip install --user strands-agents-tools
python 4.py

5. AWSにデプロイ

スクリーンショット 2025-10-04 18.29.20.png

スクリーンショット 2025-10-04 20.27.18.png

nano agent.py
# 必要なライブラリをインポート
from strands import Agent, tool
from tavily import TavilyClient
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# 先ほどと同じ「Web検索するStrandsエージェント」 ------------------------
@tool
def search(query):
    tavily = TavilyClient("tvly-xxxxxxxxxxxxxxxxxxxxxxxxxxx")
    return tavily.search(query)

agent = Agent(
    tools=[search]
)
# -----------------------------------------------------------------

# AgentCoreランタイム用のAPIサーバーを作成
app = BedrockAgentCoreApp()

# エージェント呼び出し関数を、APIサーバーのエントリーポイントに設定
@app.entrypoint
async def invoke_agent(payload, context):

    # フロントエンドで入力されたプロンプトを取得
    prompt = payload.get("prompt")

    # エージェントの応答をストリーミングで取得
    stream = agent.stream_async(prompt)
    async for event in stream:
        yield event

# APIサーバーを起動
app.run()
  • 必要なPythonパッケージの一覧ファイルを作成
nano requirements.txt
requirements.txt
strands-agents
tavily-python
bedrock-agentcore
  • スターターツールキットをセットアップ

スクリーンショット 2025-08-03 23.14.45.png

pip install --user bedrock-agentcore-starter-toolkit
agentcore configure
  • ウィザードは以下のように進める
Entrypoint: agent.py

(Memoryの前まで、途中すべて6連続でEnter)

Your choice: s
  • AgentCoreランタイムにデプロイ
agentcore deploy
  • Agent ARNをメモしておく

6. フロントエンドからアクセス

スクリーンショット 2025-10-04 22.25.42.png

nano frontend.py
frontend.py
# 必要なライブラリをインポート
import boto3, json
import streamlit as st

# サイドバーで設定を入力
with st.sidebar:
    agent_runtime_arn = st.text_input("AgentCoreランタイムのARN")

# タイトルを描画
st.title("なんでも検索エージェント")
st.write("Strands AgentsがMCPサーバーを使って情報収集します!")

# チャットボックスを描画
if prompt := st.chat_input("メッセージを入力してね"):
    # ユーザーのプロンプトを表示
    with st.chat_message("user"):
        st.markdown(prompt)

    # エージェントの回答を表示
    with st.chat_message("assistant"):

        # AgentCoreランタイムを呼び出し
        agentcore = boto3.client('bedrock-agentcore')
        payload = json.dumps({"prompt": prompt})

        response = agentcore.invoke_agent_runtime(
            agentRuntimeArn=agent_runtime_arn,
            payload=payload.encode()
        )

        ### ここから下はストリーミングレスポンスの処理 ------------------------------------------
        container = st.container()
        text_holder = container.empty()
        buffer = ""

        # レスポンスを1行ずつチェック
        for line in response["response"].iter_lines():
            if line and line.decode("utf-8").startswith("data: "):
                data = line.decode("utf-8")[6:]

                # 文字列コンテンツの場合は無視
                if data.startswith('"') or data.startswith("'"):
                    continue

                # 読み込んだ行をJSONに変換
                event = json.loads(data)

                # ツール利用を検出
                if "event" in event and "contentBlockStart" in event["event"]:
                    if "toolUse" in event["event"]["contentBlockStart"].get("start", {}):

                        # 現在のテキストを確定
                        if buffer:
                            text_holder.markdown(buffer)
                            buffer = ""

                        # ツールステータスを表示
                        container.info("🔍 Tavily検索ツールを利用しています")
                        text_holder = container.empty()

                # テキストコンテンツを検出
                if "data" in event and isinstance(event["data"], str):
                    buffer += event["data"]
                    text_holder.markdown(buffer)

                elif "event" in event and "contentBlockDelta" in event["event"]:
                    buffer += event["event"]["contentBlockDelta"]["delta"].get("text", "")
                    text_holder.markdown(buffer)

        # 最後に残ったテキストを表示
        text_holder.markdown(buffer)
        ### ------------------------------------------------------------------------------
  • フロントエンドを起動
pip install --user streamlit
streamlit run frontend.py
  • CloudShell左上の「+」ボタンより、「ap-northeast-1」をクリックして2つ目のターミナルを起動
  • 1つ目のターミナルをブラウザから覗けるよう、以下コマンドでCloudflare Tunnelに接続する
# Cloudflare Tunnelクライアントをダウンロード
wget -O cloudflared https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64

# 実行権限を付与
chmod +x cloudflared

# Streamlitのローカルポートを、Tunnelでインターネット公開
./cloudflared tunnel --url http://localhost:8501

スクリーンショット 2026-02-05 0.40.48.png

  • 2つ目のターミナルで出力されたコマンドを少し遡ると、以下のように自分専用のURLが払い出されているため、ブラウザの新しいタブでアクセス
2025-10-26T14:10:56Z INF +--------------------------------------------------------------------------------------------+
2025-10-26T14:10:56Z INF |  Your quick Tunnel has been created! Visit it at (it may take some time to be reachable):  |
2025-10-26T14:10:56Z INF |  https://happens-bond-closer-manager.trycloudflare.com                                     |
2025-10-26T14:10:56Z INF +--------------------------------------------------------------------------------------------+
  • 先ほどメモしたAgent ARNをサイドバーに入力
  • KAGのLT会って何? などと質問してみよう!

スクリーンショット 2025-10-17 18.33.17.png

次のステップ

33
16
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
33
16

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?