以下の入門書を見ながら、Strands Agentsのハンズオンをやってみます。
AIエージェント開発 / 運用入門 [生成AI深掘りガイド]
事前準備
段階的に必要になっていくものも、一括で設定しています。
-
venvの構築
pyenv install 3.12.10 --skip-existing PYENV_VERSION=3.12.10 python -m venv .venv source .venv/bin/activate pip install -r requirements.txtrequirements.txtstrands-agents==1.6.0 feedparser==6.0.11 streamlit==1.49.1 openai>=1.52.0 -
.envの定義.envOPENAI_API_KEY=<your api key> (*1) OPENAI_MODEL_ID=gpt-4o-mini # optional override OPENAI_BASE_URL=https://api.openai.com/v1 # optional override MCP_SERVERS=[{"command":"npx","args":["jgrants-mcp"]}] (*2)
*1) 今回はOpenAIのモデルを利用するため、OpenAIのAPI Keyが必要
*2) 今回はMCPサーバー(ツール)として、JGrantsを利用
Strands Agents入門
3行でエージェント実装
なんとびっくり、3行でAIエージェントが実装できるそうです。
strands_sample.py
from dotenv import load_dotenv
from strands import Agent
# .envから環境変数の読み込み
load_dotenv()
# エージェントの作成＆実行
agent = Agent("us.anthropic.claude-haiku-4-5-20251001-v1:0")
agent("Strandsとはどういう意味？")
この時点では、AWSのBedrockを利用するため、以下の環境設定が必要となります。
.env
AWS_ACCESS_KEY_ID=<your key>
AWS_SECRET_ACCESS_KEY=<your key>
AWS_DEFAULT_REGION=us-west-2
実行結果は以下のようになります。
$ python ./strands_sample.py
# Strands の意味
**Strands** は複数の意味を持ちます：
## 1. **最も一般的な意味**
- **股、ひも、糸、束** - 複数の細い線状のもの
- "strands of hair"（髪の毛）
- "strands of DNA"（DNA鎖）
## 2. **ビーチ用語**
- **浜辺、海岸** - 特に詩的または文学的な表現
- "walk along the strands"（浜辺を歩く）
## 3. **劇場用語**
- **ロンドンの劇場街** - Strand Theatre など
## 4. **比喩的な意味**
- **要素、筋道、つながり**
- "strands of a story"（物語の筋）
- "different strands of thought"（異なる思考の流れ）
## 5. **最近のトレンド**
- **NYTimes Strands** - ニューヨークタイムズの言葉探しゲーム（2023年に開始）
文脈によって意味が変わるので、前後の文章から判断する必要があります。何か特定の文脈での意味でしたら、教えてください！
モデルを変更してみる
弊社ではOpenAIに定額課金しているため、OpenAIのモデルを利用しようと思います。
strands_sample.py
import os
from dotenv import load_dotenv
from strands import Agent
from strands.models.openai import OpenAIModel
load_dotenv()
model = OpenAIModel(
model_id="gpt-4o-mini",
client_args={"api_key": os.getenv("OPENAI_API_KEY")},
)
agent = Agent(model)
print(agent("Strandsとはどういう意味？"))
このタイミングでBedrockの利用を行わないため、OpenAI向けのKeyのみでOKとなります。
.env
OPENAI_API_KEY=<your key>
実行結果は以下となります。
$ python ./strands_sample.py
「Strands」という言葉は、いくつかの意味がありますが、一般的には以下のような意味で使われます。
1. **糸や繊維**：物理的な形状として、細長いものや糸のようなものを指します。例えば、髪の毛の一房や、織物の糸などです。
2. **要素や側面**：比喩的に、何かの構成要素や側面を指すことがあります。たとえば、物語や議論の「筋」や「テーマ」の部分を指すときに使われます。
3. **生物学的な文脈**：DNAやRNAの「鎖」を指すこともあります。
文脈によって意味が変わるため、具体的な文脈を考慮することが重要です。「Strands」という言葉は、いくつかの意味がありますが、一般的には以下のような意味で使われます。
1. **糸や繊維**：物理的な形状として、細長いものや糸のようなものを指します。例えば、髪の毛の一房や、織物の糸などです。
2. **要素や側面**：比喩的に、何かの構成要素や側面を指すことがあります。たとえば、物語や議論の「筋」や「テーマ」の部分を指すときに使われます。
3. **生物学的な文脈**：DNAやRNAの「鎖」を指すこともあります。
文脈によって意味が変わるため、具体的な文脈を考慮することが重要です。
モデルが変わると文面も大分違いますね。
ツールを渡してみる
入門書で実装しているAWSのアップデート情報を取得する処理をツールとしてエージェントに渡します。
strands_sample.py
import feedparser # 追加
# エージェント向けのツールを定義
@tool
def get_aws_updates(service_name: str) -> list:
feed = feedparser.parse("https://aws.amazon.com/about-aws/whats-new/recent/feed/")
result = []
for entry in feed.entries:
if service_name.lower() in entry.title.lower():
result.append({
"published": entry.get("published", "N/A"),
"summary": entry.get("summary", "")
})
if len(result) >= 3:
break
return result
定義したツールをエージェントに渡します。
strands_sample.py
agent = Agent(
model=model
tools=[get_aws_updates]
)
MCP編
次は、MCPサーバーで提供されるツールをエージェントに渡してみます。
今回はお手軽に利用できる @tachibanayu24さん のJGrants(*3)のMCPサーバーをお借りします。
*3) デジタル庁が運営する、国や地用自治体の補助金をオンラインで申請できるシステム
strands_sample.py
from mcp.client.stdio import StdioServerParameters, stdio_client
# MCPサーバーを定義
server = StdioServerParameters(command="npx", args=["jgrants-mcp"])
# MCPクライアントを実行
client = MCPClient(lambda: stdio_client(server))
client.start()
# MCPサーバーから利用可能なツール情報を取得
tools = client.list_tools_sync()
MCPサーバーから取得したツール情報をエージェントに渡します。
strands_sample.py
agent = Agent(model=model, tools=list(tools))
StreamlitでGUIを簡易実装してツールの使い分けを確認してみる
入門書でも紹介されているStreamlit(*4)を使ってGUIを簡易実装しつつ、
これまで実装したものを少し整理しました。
*4) PythonでWebアプリケーションを簡単に開発できるオープンソースのフレームワーク
strands_sample.py
strands_sample.py
import asyncio
import json
import os
import shlex
from contextlib import suppress
import feedparser
import streamlit as st
from dotenv import load_dotenv
from mcp.client.stdio import StdioServerParameters, stdio_client
from strands import Agent, tool
from strands.models.openai import OpenAIModel
from strands.tools.mcp import MCPClient
# /** Tools **/
@tool
def get_aws_updates(service_name: str) -> list:
feed = feedparser.parse("https://aws.amazon.com/about-aws/whats-new/recent/feed/")
result = []
for entry in feed.entries:
if service_name.lower() in entry.title.lower():
result.append({
"published": entry.get("published", "N/A"),
"summary": entry.get("summary", "")
})
if len(result) >= 3:
break
return result
# /** Agents **/
def build_openai_model() -> OpenAIModel:
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY が設定されていません。")
model_id = os.getenv("OPENAI_MODEL_ID", "gpt-4o-mini")
client_args = {"api_key": api_key}
if base_url := os.getenv("OPENAI_BASE_URL"):
client_args["base_url"] = base_url
return OpenAIModel(client_args=client_args, model_id=model_id)
def start_mcp_clients():
server_defs = _load_mcp_servers()
clients: list[MCPClient] = []
tools = []
connected_labels: list[str] = []
failed_labels: list[str] = []
for server_def in server_defs:
label = _format_server_label(server_def)
server = StdioServerParameters(
command=server_def["command"],
args=server_def.get("args") or [],
env=server_def.get("env"),
cwd=server_def.get("cwd"),
)
client = MCPClient(lambda: stdio_client(server))
try:
client.start()
page = client.list_tools_sync()
tools.extend(page)
while page.pagination_token:
page = client.list_tools_sync(page.pagination_token)
tools.extend(page)
clients.append(client)
connected_labels.append(label or server_def["command"])
except Exception as exc: # pragma: no cover - surfaced in UI
failed_labels.append(label or server_def["command"])
log_warning(f"MCPサーバーに接続できませんでした ({label}): {exc}")
with suppress(Exception):
client.stop(None, None, None)
return clients, tools, connected_labels, failed_labels
def _load_mcp_servers() -> list[dict]:
raw = os.getenv("MCP_SERVERS")
if not raw:
return []
try:
data = json.loads(raw)
except Exception as exc:
log_error(f"MCP_SERVERS をパースできませんでした: {exc}")
return []
servers = data if isinstance(data, list) else [data]
result = []
for server in servers:
if not isinstance(server, dict):
continue
command = server.get("command")
if not command:
continue
args = server.get("args") or []
if isinstance(args, str):
args = shlex.split(args)
result.append(
{
"command": command,
"args": args,
"env": server.get("env"),
"cwd": server.get("cwd"),
}
)
return result
def _format_server_label(server_def: dict) -> str:
args_part = " ".join(server_def.get("args") or [])
return f"{server_def.get('command','')} {args_part}".strip()
def stop_mcp_clients(clients: list[MCPClient]) -> None:
for client in clients:
with suppress(Exception):
client.stop(None, None, None)
def build_agent():
model = build_openai_model()
mcp_clients, mcp_tools, connected_labels, failed_labels = start_mcp_clients()
agent = Agent(model=model, tools=[get_aws_updates, *mcp_tools])
return agent, mcp_clients, connected_labels, failed_labels
# /** GUI(Streamlit) **/
async def stream_chat(message: str, container, agent):
text_holder = container.empty()
response = ""
prompt = message.strip()
async for chunk in agent.stream_async(prompt):
if isinstance(chunk, dict):
event = chunk.get("event", {})
# Debug: show raw event blocks in sidebar
st.sidebar.write("event:", event)
if "contentBlockStart" in event:
tool_use = event["contentBlockStart"].get("start", {}).get("toolUse", {})
tool_name = tool_use.get("name")
if not tool_name:
continue
if response:
text_holder.markdown(response)
response = ""
container.info(f"🔧 {tool_name} ツールを実行中...")
text_holder = container.empty()
if text := chunk.get("data"):
response += text
text_holder.markdown(response)
def ensure_agent_session():
if "agent_session" in st.session_state:
return st.session_state["agent_session"]
try:
agent, mcp_clients, connected_labels, failed_labels = build_agent()
session = {
"agent": agent,
"mcp_clients": mcp_clients,
"connected_labels": connected_labels,
"failed_labels": failed_labels,
}
st.session_state["agent_session"] = session
return session
except Exception as exc:
log_error(f"エージェント初期化に失敗しました: {exc}")
return None
# /** Logger **/
def log_error(message: str) -> None:
st.error(message)
def log_warning(message: str) -> None:
st.warning(message)
def log_success(message: str) -> None:
st.success(message)
def log_info(message: str) -> None:
st.info(message)
# /** Main logic **/
def main():
st.title("Sample AI Agents (OpenAI + MCP)")
session = ensure_agent_session()
connected_labels = session["connected_labels"] if session else []
failed_labels = session["failed_labels"] if session else []
if connected_labels:
log_success("接続済みMCPサーバー: " + ", ".join(connected_labels))
if failed_labels:
log_warning("接続に失敗したMCPサーバー: " + ", ".join(failed_labels))
if session:
tool_names = [getattr(tool, "tool_name", "") for tool in session["agent"].tool_registry.registry.values()]
if tool_names:
log_info("利用可能なツール: " + ", ".join(sorted(filter(None, tool_names))))
user_message = st.text_area("メッセージを入力してください", height=160)
if st.button("送信"):
if user_message:
with st.spinner("回答用意中..."):
container = st.container()
session = ensure_agent_session()
if session:
agent = session["agent"]
asyncio.run(stream_chat(user_message, container, agent))
if __name__ == "__main__":
load_dotenv()
main()
整理後のサンプルコードではMCPサーバーも環境変数から取得するようにしているため、以下の定義が必要
.env
OPENAI_API_KEY=<your key>
MCP_SERVERS=[{"command":"npx","args":["jgrants-mcp"]}]
以下のコマンドで起動します。
streamlit run strands_sample.py
以下が実行結果となり、エージェントが質問によってツールを使い分けているのがわかります。
次回はMastraでのハンズオンを実施してみようと思います。
実行結果：独自ツール(get_aws_updates)を利用
実行結果：MCPツール(list_subsidies)を利用
