Streamlitでアプリを作って実感するStrands Agentsのいいところ

Posted at 2025-05-20

Strands Agentsにもうちょっと踏み込んでみたいと思い、Streamlitのチャットを作りました。

この記事の前にこのあたりを先に読まれるとスムーズかと思います。

事前準備

ライブラリーをインストールしましょう。

uv add strands-agents streamlit nest-asyncio

1. 最低限の機能で実装する

ほとんどStrands Agentsの良さは発揮されてませんが、一番最低限なチャットはこうなります。

1_simple.py

import boto3
import streamlit as st
from strands import Agent
from strands.models import BedrockModel

st.title("Strands agent")


if prompt := st.chat_input():
    with st.chat_message("user"):
        st.write(prompt)

    agent = Agent(
        model=BedrockModel(
            model_id="us.amazon.nova-pro-v1:0",
            boto_session=boto3.Session(region_name="us-east-1"),
        ),
        callback_handler=None
    )

    response = agent(prompt=prompt)

    st.write(response.message["content"][0]["text"])

callback_handlerが未指定の場合は、デフォルトでPrintingCallbackHandlerというコールバックハンドラーが指定されます。（PrintingCallbackHandlerは標準出力にBedrockの返答を出力します。）
Streamlitで画面上に出力するので標準出力への出力を抑制するためにNoneを指定します。

2. ストリームでレスポンスを出力する

次はストリームで出力する方法です。これもほとんどConverse APIを使う場合と変わらないです。

Streamlitの非同期処理とバッティングするのか（？）、nest_asyncioが必要でした。

2_streaming.py

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            callback_handler=None
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

3. 会話履歴を保持する

ここまでの実装では、毎回新規の会話が行われますので、過去の会話内容は反映されません。

エージェントの実行後、その会話のやり取りはagent.messagesで取得できるので、これをst.session_stateで管理します。

3_conversation_history.py

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            with st.chat_message(message["role"]):
                st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

4. 会話の最大数を管理する

このあたりからStrands Agentsの機能が活用されていきます。

会話の最大数を決めて、自動的に過去の会話を自動的に忘れることができます。

Agentのconversation_managerパラメーターにSlidingWindowConversationManagerを指定します。

window_sizeを奇数にすると、userメッセージだけ削除され、次のリクエストで「userから始まってないよエラー」になります。なので、偶数である必要がありそうです。

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.agent.conversation_manager import SlidingWindowConversationManager
from strands.models import BedrockModel
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            with st.chat_message(message["role"]):
                st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            conversation_manager=SlidingWindowConversationManager(window_size=4)
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

5. ツールを使う（ビルドインツール）

このあたりから「おっ！」と思う感じになってきます。

Strands Agentsには、豊富なビルドインツールが提供されています。

なかなか面白そうなツールが盛り沢山です。

Strands Agentsとは別のライブラリーとして提供されていますので、追加でインストールします。

uv add strands-agents-tools

ツールを呼び出す場合、いわゆる「イベントループ」の形でwhileループを組む必要があります。

Strands Agentsを使うと、 このイベントループ処理を時前で実装する必要はありません！ （すごい！）

もうこれだけでStrands Agentsの採用決定です。

import asyncio
import os

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages
from strands_tools import shell

nest_asyncio.apply()

os.environ["DEV"] = "true"

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            tools=[shell],
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

6. ツールを使う（MCPツール）

Strands AgentsはMCPにも対応しています。

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from mcp import StdioServerParameters, stdio_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        stdio_mcp_client = MCPClient(
            lambda: stdio_client(
                StdioServerParameters(
                    command="uvx",
                    args=["awslabs.aws-documentation-mcp-server@latest"],
                    env={"FASTMCP_LOG_LEVEL": "ERROR"},
                )
            )
        )

        with stdio_mcp_client:
            tools = stdio_mcp_client.list_tools_sync()

            agent = Agent(
                model=BedrockModel(
                    model_id="us.amazon.nova-pro-v1:0",
                    boto_session=boto3.Session(region_name="us-east-1"),
                ),
                messages=st.session_state.messages if "messages" in st.session_state else [],
                callback_handler=None,
                tools=tools,
            )

            agent_stream = agent.stream_async(prompt=prompt)

            st.write_stream(streaming(agent_stream))

            st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

7. トレースもとれる

Strands Agentsではオブザーバビリティ機能も用意されています。トレース情報はOpenTelemetryの形式で対応しているとのことです。

みんな大好きLangfuseはすでにドキュメントに記載があります。はやい！

Arize Phoenixを使う場合は、以下のように環境変数``を設定するだけです。

import asyncio
import os

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages
from strands_tools import shell

nest_asyncio.apply()

os.environ["DEV"] = "true"
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:6006/v1/traces"

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            tools=[shell],
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

だがしかし、日本語は正しく表示されません。

ここでプルリク上げてますので、マージされるのをお待ち下さい。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up