AgentCoreまつり、開催中です。
@har1101 さんが投稿されたこちらの記事を読みました。
なにか他の手がないのかなと思って調べたら、いい感じのものを発見しました！
参考にしたのはこちらのサンプルです。
AgentCore Runtimeから外部のAPIを呼ぶ際のAPIキー
OpenAIのAPIを呼ぶアプリを作ります。
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from openai import OpenAI
app = BedrockAgentCoreApp()
@app.entrypoint
async def invoke(payload: dict):
prompt = payload.get("prompt")
client = OpenAI()
response = client.responses.create(model="gpt-4o", input=prompt)
return response
if __name__ == "__main__":
app.run()
さて、OpenAIのAPIキーはどうしましょう。
ここでAgentCore Identityの登場です。
マネジメントコンソールでAPIキーを登録
画面をポチポチします。
APIキーの取得方法
AgentCore Identityに登録したAPIキーは、AgentCore SDKを使うとシュシュっと取得するためのrequires_api_keyというデコレーター関数が用意されています。
インポートします。
from bedrock_agentcore.identity.auth import requires_api_key
requires_api_keyデコレーターをつけた関数を用意します。provider_nameはマネジメントコンソールに登録した際の「Name」です。
@requires_api_key(provider_name="OPENAI_API_KEY")
async def need_api_key(*, api_key: str):
global OPENAI_API_KEY
print("received api key for async func")
OPENAI_API_KEY = api_key
この関数を呼び出すと、APIキーが取得できます。すごい！
await need_api_key(api_key="")
組み合わせると、こんな感じになります。
import os
from bedrock_agentcore.identity.auth import requires_api_key
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from openai import OpenAI
os.environ["AWS_REGION"] = "us-west-2"
app = BedrockAgentCoreApp()
OPENAI_API_KEY = ""
@requires_api_key(provider_name="OPENAI_API_KEY")
async def need_api_key(*, api_key: str):
global OPENAI_API_KEY
print("received api key for async func")
OPENAI_API_KEY = api_key
@app.entrypoint
async def invoke(payload: dict):
prompt = payload.get("prompt")
await need_api_key(api_key="")
client = OpenAI(api_key=OPENAI_API_KEY)
response = client.responses.create(model="gpt-4o", input=prompt)
return response
if __name__ == "__main__":
app.run()
リクエストのたびに取得しなくてもいいので、環境変数にセットしても良いと思います。
@requires_api_key(provider_name="OPENAI_API_KEY")
async def need_api_key(*, api_key: str):
print("received api key for async func")
os.environ["OPENAI_API_KEY"] = api_key
...
if not os.environ.get("OPENAI_API_KEY"):
await need_api_key(api_key="")
AgentCore RuntimeからAgentCore Gatewayを呼ぶ際のBearerトークン
感動するのはまだ早いです。
AgentCore Gatewayを呼び出す際のBearerトークンもシュシュっと取得できます。
先日投稿したこちらの記事の内容で、Tavily検索をAgentCore Gatewayにデプロイした状態から始めます。
マネジメントコンソールでIdentityを登録
Cognitoの画面で以下の情報をメモります。
-
クライアントID
-
クライアントシークレット
-
カスタムスコープ
AgentCore Gatewayの画面で以下の情報をメモります。
-
Discovery URL
これらの情報をAgentCore Identityに登録します。
「Add OAuth client / API key」のボタンメニューを開いた後、「Add OAuth Client」を選択します。
Providerを「Custom provider」にし、「Discovery URL」を選択した状態で各項目に入力します。
トークンの取得方法
APIキー取得とほとんど同じですが、requires_access_tokenデコレーター関数を使用します。
from bedrock_agentcore.identity.auth import requires_access_token
@requires_access_token(
provider_name="Tavily-gateway-quick-start-726b0d", # AgentCore IdentityのName
scopes=["Tavily-gateway-quick-start-726b0d/genesis-gateway:invoke"], # Cognitoの画面から取得した値
auth_flow="M2M",
)
async def need_api_key(*, access_token: str):
global ACCESS_TOKEN
print("received api key for async func")
ACCESS_TOKEN = access_token
あとはMCPクライアントから呼ぶだけです。
async def main():
await need_api_key(access_token="")
mcp_url = "https://tavily-gateway-quick-start-726b0d-yujo4vy0pe.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
async with streamablehttp_client(
mcp_url, headers, timeout=120, terminate_on_close=False
) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tool_result = await session.list_tools()
print(tool_result)
スマートですね！
気付いた点
AgentCore Gatewayを作成した際に自動で作成されるCognitoですが、M2Mというタイプで作成されるようです。
これは、人を認証するのではなくマシン（今回のケースではAgentCore Runtime）を認証するという仕組みのようです。
で、このM2Mの場合、Cognitoの料金体系が異なります。
アプリケーションクライアントごとに最低6ドル/月と、トークンを発行するたびにお金が掛かりそうです。
Cognitoって限りなく0円に近いサービスと思って使っていると、ちょっとびっくりします。