こんにちは、ふくちです。
以下ブログでEventBridge × AgentCore Runtimeのスケジュール駆動Ambient Agentを作ったのですが、エージェントの機能を拡張していくと、いくつかのエラーに遭遇しました。
このエラーの事象・原因・対処法などをまとめておきます。
作ったエージェントについて
一言で言うと、Connpassのイベント情報収集エージェントです。
同僚や後輩にIT関連のイベント(とりわけJAWS-UGのイベント)を紹介する機会が増えてきたのですが、「自分で検索する→日時や内容を纏める→LINEに投稿する」の流れが大変なので、代わりにやってくれるAIエージェントを作ってみたという感じです。
- スケジュール駆動でエージェントが動く
- エージェントがConnpass API v2用ツールを使って情報収集
- LINE Bot MCPサーバーを使い通知する
という流れです。
イメージはこんな感じ(今はJAWS-UG以外のイベントも対象にしています)。
この時、LINE Bot MCPサーバーを使うためにLINE関連のシークレットやトークンを用いるのですが、ここでStrands AgentsとAgentCore Identityを用いて管理していました。
その際にエラーがいくつか発生したので、備忘録として残しておきます。
Amazon Bedrock AgentCore で「Workload access token has not been set」エラーが発生する場合
事象
AgentCore Runtimeにデプロイしたエージェントにおいて、 @requires_api_key デコレーターを使用してAgentCore Identityから認証情報を取得しようとすると、以下のエラーが発生することがあります。
ERROR: Workload access token has not been set. If invoking agent runtime via SIGV4 inbound auth, please specify the X-Amzn-Bedrock-AgentCore-Runtime-User-Id header and retry.
For details, see - https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-oauth.html
原因
原因は以下の2点です。
1. SIGV4 認証時は RuntimeUserId パラメータが必須
EventBridge SchedulerやAWS SDKからinvokeAgentRuntime APIを呼び出す際、IAM認証(SIGV4)を使用している場合は RuntimeUserIdパラメータを明示的に指定する必要があります。
RuntimeUserId が指定されていない
→AgentCore Runtime は Workload Access Token を生成できない
→@requires_api_key デコレーターで処理が上手くいかない
→エラーになる
という流れです。
これは以下記事でも触れている内容なので、もう少し詳しく知りたい方はこちらをご覧ください。
2. @requires_api_keyは@app.entrypoint内でのみ動作する
@requires_api_keyデコレーターは、リクエスト時にRuntimeが提供するWorkload Access Tokenを必要とします。
そのため、@app.entrypointでデコレートされた関数内においてのみ使用することができます。
AgentCore Runtimeの非同期実行を用いる際は更に注意が必要で、本処理をバックグラウンドで行う前にAgentCore Identityからシークレットを取得する必要があります。
すなわちasyncio.create_task()で起動された非同期タスク内では
@app.entrypointのコンテキスト外となる
→Workload Access Tokenが利用できない
→結果として@requires_api_keyが正しく処理されない
という流れになります。
参考：以下はダメな例。
async def get_line_credentials():
"""AgentCore IdentityからLINE Bot認証情報を取得"""
credentials = {}
@requires_api_key(provider_name="LINE_BOT_TOKEN")
async def _get_token(*, api_key: str):
credentials["token"] = api_key
await _get_token() # ここで Workload Access Token が必要
return credentials
async def create_line_mcp_client():
"""LINE Bot MCPクライアントを作成"""
credentials = await get_line_credentials() # @requires_api_key を呼び出す
# ... MCPクライアント作成処理
async def _background_run(task_id, payload, context):
"""バックグラウンド処理"""
# NG: asyncio.create_task() で起動されたタスク内なので
# Workload Access Token が利用できずエラーになる
line_mcp_client = await create_line_mcp_client()
# ...
@app.entrypoint
async def main(payload, context=None):
if payload.get("action") == "start":
task_id = app.add_async_task("agent_job", {"job_id": payload.get("job_id")})
# NG: ここで asyncio.create_task() を使うと、
# _background_run 内の @requires_api_key は entrypoint コンテキスト外になる
asyncio.create_task(_background_run(task_id, payload, context))
return {"status": "started", "task_id": task_id}
return {"status": "noop"}
対処法
Step 1: EventBridge SchedulerにRuntimeUserIdを追加
以下はCDKでEventBridge Schedulerを立てている時の実装例です。
inputのところでRuntimeUserIdを加えてあげてください。
// EventBridge Scheduler
new scheduler.CfnSchedule(this, 'AgentSchedule', {
name: scheduleName,
groupName: scheduleGroupName,
scheduleExpression: atExpression,
scheduleExpressionTimezone: 'Asia/Tokyo',
flexibleTimeWindow: { mode: 'OFF' },
state: 'ENABLED',
target: {
arn: 'arn:aws:scheduler:::aws-sdk:bedrockagentcore:invokeAgentRuntime',
roleArn: schedulerRole.roleArn,
input: JSON.stringify({
AgentRuntimeArn: runtime.agentRuntimeArn,
RuntimeUserId: 'scheduler-agent', // これを追加
Payload: payload,
}),
},
});
Step 2: IAM権限にInvokeAgentRuntimeForUserを追加
以下ドキュメントによると、権限を追加する必要があります。
Invoking InvokeAgentRuntime with the X-Amzn-Bedrock-AgentCore-Runtime-User-Id header will require a new IAM action: bedrock-agentcore:InvokeAgentRuntimeForUser, in addition to the existing bedrock-agentcore:InvokeAgentRuntime action.
schedulerRole.addToPolicy(new iam.PolicyStatement({
effect: iam.Effect.ALLOW,
actions: [
'bedrock-agentcore:InvokeAgentRuntime',
'bedrock-agentcore:InvokeAgentRuntimeForUser', // これが必要
],
resources: [
runtime.agentRuntimeArn,
`${runtime.agentRuntimeArn}/runtime-endpoint/DEFAULT`,
],
}));
Step 3: @app.entrypoint内で認証情報を取得
先程の失敗例では、@app.entrypoint外で@require_api_keyを取得しようとしていました。
それをここでは修正し、@app.entrypoint内でシークレット取得処理を行い、それを後続のバックグラウンドタスクに渡すという形を取っています。
from bedrock_agentcore.identity.auth import requires_api_key
# 認証情報取得関数
async def get_line_credentials():
"""AgentCore IdentityからLINE Bot認証情報を取得"""
credentials = {}
@requires_api_key(provider_name="LINE_BOT_TOKEN")
async def _get_token(*, api_key: str):
credentials["token"] = api_key
@requires_api_key(provider_name="LINE_BOT_USER_ID")
async def _get_user_id(*, api_key: str):
credentials["user_id"] = api_key
await _get_token()
await _get_user_id()
return credentials
# 認証情報を使ってMCPクライアントを作成する関数
def create_line_mcp_client_with_credentials(credentials: dict = None):
"""事前に取得した認証情報からLINE Bot MCPクライアントを作成"""
if not credentials:
return None
token = credentials.get("token")
user_id = credentials.get("user_id")
if not token:
return None
env = {"CHANNEL_ACCESS_TOKEN": token}
if user_id:
env["DESTINATION_USER_ID"] = user_id
return MCPClient(lambda: stdio_client(
StdioServerParameters(
command="npx",
args=["-y", "@line/line-bot-mcp-server"],
env=env
)
))
# バックグラウンド処理(認証情報を引数で受け取る)
async def _background_run(task_id: int, payload: Dict[str, Any], context, line_credentials: dict = None):
# 引数で受け取った認証情報を使用
line_mcp_client = create_line_mcp_client_with_credentials(line_credentials)
# ... 以下処理
# エントリポイント(@app.entrypoint 内で認証情報を取得)
@app.entrypoint
async def main(payload: Dict[str, Any], context=None):
if payload.get("action") == "start":
# entrypoint 内で認証情報を取得(ここでは Workload Access Token が有効)
try:
line_credentials = await get_line_credentials()
log.info("[LINE] AgentCore Identityから認証情報を取得しました")
except Exception as e:
log.error("[LINE] AgentCore Identityからの認証情報取得に失敗: %s", e)
line_credentials = None
task_id = app.add_async_task("agent_job", {"job_id": payload.get("job_id")})
# 認証情報を _background_run に渡す
asyncio.create_task(_background_run(task_id, payload, context, line_credentials))
return {"status": "started", "task_id": task_id}
return {"status": "noop"}
ここまでのポイントまとめ
ざっくりまとめると以下のとおりです。
| 項目 | 説明 |
|---|---|
RuntimeUserId |
SIGV4 認証時に Workload Access Token を生成するために必須 |
InvokeAgentRuntimeForUser |
RuntimeUserId 指定時に必要な IAM 権限 |
@requires_api_key の呼び出し位置 |
必ず @app.entrypoint 内で呼び出す |
| 非同期タスクへの認証情報の受け渡し | 関数引数として明示的に渡す |
Strands Agents SDK の MCPAgentTool で 'MCPAgentTool' object has no attribute 'name' エラーが発生する
事象
Strands Agents SDKでMCPクライアントからツール一覧を取得しツール名をログ出力しようとすると、以下のエラーが発生することがあります。
line_tools = line_mcp_client.list_tools_sync()
log.info("Tools: %s", [t.name for t in line_tools]) # エラー発生
AttributeError: 'MCPAgentTool' object has no attribute 'name'
原因
これはStrandsの仕様が原因です。
まず前提としてMCPの仕様では、ツールはnameプロパティを持っています。
{
"name": "get_weather",
"description": "Get current weather...",
"inputSchema": { ... }
}
しかしStrandsでは、MCPAgentToolクラスの仕様上、tool_name or mcp_tool.nameのどちらかを使う必要があります。
StrandsのMCPAgentToolクラスは、MCPツールをエージェントフレームワークに適応させるためのラッパーです。
| プロパティ/メソッド | 説明 |
|---|---|
tool_name |
エージェントフレームワーク側で使用する名前(オーバーライド可能) |
mcp_tool.name |
MCPサーバーとの通信に使用する元の名前 |
tool_spec |
MCP仕様をエージェントのToolSpec形式に変換した辞書 |
tool_type |
常にpythonを返す |
stream() |
MCPサーバーへのツール呼び出しを非同期で実行 |
# StrandsのMCPAgentTool実装(簡略版)
class MCPAgentTool:
"""MCP ツールをエージェントフレームワークに適応させるアダプタ"""
def __init__(self, mcp_tool, mcp_client, name_override=None, timeout=None):
self.mcp_tool = mcp_tool # 元の MCP ツールオブジェクト
self.mcp_client = mcp_client # MCP クライアント接続
self.timeout = timeout # タイムアウト設定（オプション）
# name_override があればそれを使用、なければ元の名前を使用
self._agent_tool_name = name_override or mcp_tool.name
@property
def tool_name(self) -> str:
"""ツール名を取得する。
Returns:
str: ツールがエージェントに提示する名前(曖昧さが解消されている場合があります)
"""
return self._agent_tool_name
@property
def tool_spec(self) -> dict:
"""MCP仕様をエージェントのToolSpec形式に変換"""
return {
"name": self.tool_name, # ← ここでtool_nameを使用
"description": self.mcp_tool.description,
"inputSchema": self.mcp_tool.inputSchema,
# ...
}
async def stream(self, tool_use):
"""MCPサーバーへのツール呼び出しを実行"""
result = await self.mcp_client.call_tool(
self.mcp_tool.name, # ← サーバー通信では元のmcp_tool.nameを使用
tool_use["input"]
)
# ...
全実装を確認したい場合は以下をご覧ください。
なぜtool_nameとmcp_tool.nameの2つがあるのか
理由は、Disambiguation(曖昧性排除) のためと考えられます。
- 複数の MCP サーバーが同名のツール (例:
execute,run) を提供する場合がある - そこで、以下のように区別し使い分ける
- エージェント側では名前空間プレフィックスを付けて区別する
- サーバー通信時は元の
mcp_tool.nameを使用する
# 名前空間プレフィックスを指定してツールを取得
atlassian_tools = atlassian_client.list_tools_sync(prefix="atlassian")
github_tools = github_client.list_tools_sync(prefix="github")
# これで同名ツールの衝突を回避できる
all_tools = atlassian_tools + github_tools
# → 1つ目のMCPサーバーにおける"search" が "atlassian_search" になる
# → 2つ目のMCPサーバーにおける"search" が "github_search" になる
裏側の処理については、以下の公式実装をご参照ください。
対処法
問題が起こっているのは、エージェント側の処理です。
エージェント側でMCPクライアント作成時に'MCPAgentTool' object has no attribute 'name'というエラーが出ています。
なのでここでは、nameではなくtool_nameを使用するべきです。
with line_mcp_client:
line_tools = line_mcp_client.list_tools_sync()
# MCPAgentTool は tool_name 属性を持つ
tool_names = [getattr(t, 'tool_name', getattr(t, '__name__', str(t))) for t in line_tools]
log.info("[SearchEventsAgent] LINE MCP tools loaded: %s", tool_names)
mcp_agent = Agent(
model=orchestrator_model,
tools=line_tools,
system_prompt=SYSTEM_PROMPT,
session_manager=session_manager
)
ここまでのポイントまとめ
| レイヤー | プロパティ名 | 実装時の注意 |
|---|---|---|
| MCP | name |
仕様で定義 |
MCPAgentToolクラス |
tool_name |
エージェントの処理においてはこちらを使う (曖昧性排除のため) |
MCPAgentToolクラス |
mcp_tool.name |
MCPサーバーとの通信時はこちらを使う (MCPサーバー側が定義するツール名を使うため) |
まとめ
- Runtime内のエントリー処理中のみ、AgentCore Identityのシークレット取得が可能！
- StrandsでMCPツールを使用する際は
tool_nameを使おう！