AI SDK useChatのバックエンドをStrands Agentsにする方法！（バックエンドはNext.jsじゃなくてもいい！！）

先週投稿した記事が沢山のイイネを頂きました！皆様、ありがとうございます！！

いい感じにできたのですが、次の欲望が出てきました。

せっかくBedrock AgentsにデプロイするならAIエージェントはStrands Agentsで作りたい

フロントエンドはAI SDKのuseChatを使いつつ、バックエンドをPythonにする方法を紹介します。

useChatの仕様が公開されている

AI SDKのuseChatで使われている通信方式は、「Stream Protocols」としてドキュメントが公開されています。

この仕様にそってバックエンドを構築できれば、Strands Agentsで実現できそうですね。

やってみよう

フロントエンドのソースは前回のソースと同じものを使います。

バックエンドをPythonで構築していきます。まずはPythonプロジェクトを作成します。

uv init backend --python 3.13

FastAPIとStrands Agentsをインストールします。

cd backend
uv add fastapi strands-agents

main.pyに実装をしていきます。FastAPIの最低限のコードを記述します。
前回同様、フロントエンドと別サーバー扱いなのでCORSの設定が必要です。

main.py

import json

from fastapi import FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from strands.agent import Agent
from strands.models import BedrockModel

class Request(BaseModel):
    id: str
    trigger: str
    messages: list[dict]
    model: str # フロントエンドで追加した項目
    reasoning: str # フロントエンドで追加した項目

app = FastAPI()

origins = [
    "http://localhost:3000",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


async def stream_text(request, protocol):
    pass


@app.post("/invocations")
def invocations(request: Request, protocol: str = Query("data")):
    response = StreamingResponse(stream_text(request, protocol))
    response.headers["x-vercel-ai-ui-message-stream"] = "v1"
    return response

Requestはフロントエンドから送られるリクエストボディの型定義です。Chromeでリクエストを投げながら確認しました。

stream_textの実装を進めていきます。

ドキュメントによると、「Text Stream Protocol」と「Data Stream Protocol」があります。実際の通信を確認すると「Data Stream Protocol」が使われていたのでこちらで実装を進めます。

Data Stream Protocolの仕様として、

• メッセージの開始、終了
• テキストの開始、差分（デルタ）、終了
• 思考の開始、差分（デルタ）、終了
• ツールのインプット、アウトプット
• ストリームの終了

などのイベントが発生するたびにメッセージをServer-Sent Events (SSE)で返却する形になります。

メッセージの開始イベントの例

data: {"type":"start","messageId":"..."}

どう実装していいかわからず試行錯誤していたのですが以下の実装でうまくいきました。

• 値はyieldで返却する
• 「data: 」を含む 文字列 を返却する
• 各メッセージのあとに２つ改行が必要（\n\n）

先程のstream_text関数に固定でイベントを返却して動作をチェックします。

async def stream_text(request, protocol):
    yield 'data: {"type":"start","messageId":"message001"}\n\n'
    yield 'data: {"type":"text-start","id":"message001"}\n\n'
    yield 'data: {"type":"text-delta","id":"message001","delta":"Hello"}\n\n'
    yield 'data: {"type":"text-delta","id":"message001","delta":"こんにちは"}\n\n'
    yield 'data: {"type":"text-end","id":"message001"}\n\n'
    yield 'data: {"type":"finish"}\n\n'
    yield "data: [DONE]\n\n"

実装方法がわかったのでStrands Agentsと組み合わせましょう。

Strands Agentsのイベントの形はこちらにあるのですが、具体的にどんな内容が含まれるのかは動作させて確認するのが早いでしょう。

Strands Agentsのagentを作成します。

async def stream_text(request: Request, protocol: str):
    model_id = request.model
    reasoning = request.reasoning
    prompt = request.messages[0]["parts"][0]["text"]

    model = BedrockModel(
        model_id=model_id,
        region_name="ap-northeast-1",
        additional_request_fields={"reasoning_effort": reasoning},
    )
    agent = Agent(
        model,
        system_prompt="You are a helpful assistant that can answer questions and help with tasks",
        callback_handler=None,
    )

Strands Agentsのイベントとして「テキストコンテンツを開始します」や「思考コンテンツを開始します」というものがないので、contentBlockIndexが変わったら開始したことにするという実装にしてます。

    content_block_index = -1
    type = ""
    id = ""

    async for event in agent.stream_async(prompt):
        message_id = request.id
        if "event" in event:
            if "messageStart" in event["event"]:
                data = {"type": "start", "messageId": message_id}
                yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

            if "contentBlockDelta" in event["event"]:
                now_index = event["event"]["contentBlockDelta"]["contentBlockIndex"]

                if "text" in event["event"]["contentBlockDelta"]["delta"]:
                    if now_index > content_block_index:
                        content_block_index = now_index
                        type = "text"
                        id = f"{message_id}-{content_block_index}"

                        data = {"type": "text-start", "id": id}
                        yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

                    text = event["event"]["contentBlockDelta"]["delta"]["text"]
                    data = {"type": "text-delta", "id": id, "delta": text}
                    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

            if "contentBlockStop" in event["event"]:
                if type == "text":
                    data = {"type": "text-end", "id": id}
                    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
                if type == "reasoning":
                    data = {"type": "reasoning-end", "id": id}
                    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

    data = {"type": "finish"}
    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
    data = "[DONE]"
    yield f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

気力が尽きたのでテキストと思考コンテンツだけですが、こちらを参照ください。

これでAI SDKのバックエンドをStrands Agentsにできました！
動作させるとこんな感じです。

AgentCore SDKを使うともうちょっと簡単に

謎の「data: 」ですが、どこかで見覚えありませんか？？そうです、Bedrock AgentCoreのRuntimeのドキュメントです。

なんやねんこの「data: 」って思ってたんですが、SSEで返却するとこうなるんですかね。

先程のFastAPIの実装ではyieldのたびに「data: ...\n\n」としてましたが、AgentCore SDKを使うと勝手にやってくれます！ ありがとう！AgentCore SDK！！

というわけで、簡単に解説です。

AgentCore SDKをインストールします。

uv add bedrock-agentcore

yieldの部分は、JSONオブジェクトのままでOKです。

data = {"type": "text-start", "id": id}
yield data

他の部分もよしなに変更してください。完成品がこちらです。

uv run main.pyで起動すると、フロントエンドからアクセスできます。

AgentCore Runtimeへのデプロイは皆さんでお試しください！

AgentCore、こういう使い方も想定して開発されてるんですかね？よく考えられてる！