ECSの新機能Expressモードで、認証つきStreamlitアプリを簡単にAWSデプロイしよう！ #AI – Qiita

import asyncio
import streamlit as st
from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import stdio_client, StdioServerParameters

# フロントエンドを描画
st.title("Strands MCPエージェント")
st.text("AWSナレッジMCPサーバーを使って、Nova Premierがあなたの質問に答えます！")
prompt = st.chat_input("質問を入力")

if prompt:
    # ユーザーのプロンプトを表示
    with st.chat_message("user"):
        st.markdown(prompt)

    # エージェントの応答を表示
    with st.chat_message("assistant"):
        with st.spinner("考え中…"):

            # MCPクライアント作成
            client = MCPClient(lambda: stdio_client(
                StdioServerParameters(
                    command="uvx",
                    args=["awslabs.aws-documentation-mcp-server@latest"])
                )
            )

            with client:
                # エージェント作成
                agent = Agent(
                    model="us.amazon.nova-premier-v1:0",
                    system_prompt="思考も回答も日本語で行ってください。",
                    tools=client.list_tools_sync()
                )

                # ストリーミング表示の準備
                container = st.container()
                state = {
                    "text_holder": container.empty(),
                    "buffer": "",
                    "shown_tools": set()
                }

                # Strandsをストリーミング実行する非同期関数を定義
                async def run_stream():
                    async for event in agent.stream_async(prompt):
                        current_tool = event.get('current_tool_use', {})
                        tool_id = current_tool.get('toolUseId')
                        tool_name = current_tool.get('name')

                        # ツール実行を検出して表示
                        if tool_id and tool_name and tool_id not in state["shown_tools"]:
                            state["shown_tools"].add(tool_id)
                            if state["buffer"]:
                                state["text_holder"].markdown(state["buffer"])
                                state["buffer"] = ""
                            container.info(f"🔧 **{tool_name}** ツールを実行中...")
                            state["text_holder"] = container.empty()

                        # テキストを抽出して表示
                        if event.get('data'):
                            state["buffer"] += event['data']
                            state["text_holder"].markdown(state["buffer"] + "▌")

                    # 最終表示
                    if state["buffer"]:
                        state["text_holder"].markdown(state["buffer"])

                # 非同期関数を実行
                asyncio.run(run_stream())

FROM python:3.13

# 作業ディレクトリ
WORKDIR /app

# 依存パッケージをインストール
RUN pip install uv mcp streamlit strands-agents strands-agents-tools

# アプリ本体をコピー
COPY . /app

# 公開ポート
EXPOSE 80

# Streamlitを起動
CMD ["streamlit", "run", "app.py", "--server.port=80", "--server.address=0.0.0.0"]