ツールでの認証¶
コアコンセプト¶
多くのツールは、保護されたリソース（Googleカレンダーのユーザーデータ、Salesforceのレコードなど）にアクセスする必要があり、認証を必要とします。ADKは、さまざまな認証方法を安全に処理するためのシステムを提供します。
関与する主要なコンポーネントは以下の通りです：
AuthScheme: APIが認証情報をどのように期待するかを定義します（例：ヘッダーのAPIキーとして、OAuth 2.0のBearerトークンとして）。ADKはOpenAPI 3.0と同じタイプの認証スキームをサポートしています。各種類の認証情報の詳細については、OpenAPIドキュメント：認証を参照してください。ADKはAPIKey、HTTPBearer、OAuth2、OpenIdConnectWithConfigのような特定のクラスを使用します。AuthCredential: 認証プロセスを開始するために必要な初期情報を保持します（例：アプリケーションのOAuthクライアントID/シークレット、APIキーの値）。これには、認証情報の種類を指定するauth_type（API_KEY、OAUTH2、SERVICE_ACCOUNTなど）が含まれます。
一般的なフローは、ツールの設定時にこれらの詳細を提供することです。ADKはその後、ツールがAPI呼び出しを行う前に、初期の認証情報を利用可能なもの（アクセストークンなど）に自動的に交換しようとします。ユーザーの操作（OAuthの同意など）が必要なフローについては、エージェントクライアントアプリケーションを巻き込んだ特定の対話型プロセスがトリガーされます。
サポートされている初期認証情報の種類¶
- API_KEY: 単純なキー/値の認証用。通常、交換は不要です。
- HTTP: Basic認証（交換は非推奨/非サポート）または既に取得済みのBearerトークンを表すことができます。Bearerトークンの場合、交換は不要です。
- OAUTH2: 標準のOAuth 2.0フロー用。設定（クライアントID、シークレット、スコープ）が必要で、多くの場合、ユーザーの同意のための対話型フローをトリガーします。
- OPEN_ID_CONNECT: OpenID Connectに基づく認証用。OAuth2と同様に、多くの場合、設定とユーザーの操作が必要です。
- SERVICE_ACCOUNT: Google Cloudサービスアカウントの認証情報（JSONキーまたはApplication Default Credentials）用。通常、Bearerトークンに交換されます。
ツールでの認証設定¶
ツールの定義時に認証を設定します：
-
RestApiTool / OpenAPIToolset: 初期化時に
auth_schemeとauth_credentialを渡します。 -
GoogleApiToolSetツール: ADKには、Googleカレンダー、BigQueryなどの組み込みの1stパーティツールがあります。ツールセットの特定の方法を使用します。
-
APIHubToolset / ApplicationIntegrationToolset: API Hubで管理されているAPI/Application Integrationによって提供されるAPIが認証を必要とする場合、初期化時に
auth_schemeとauth_credentialを渡します。
警告
アクセストークンや特にリフレッシュトークンのような機密性の高い認証情報をセッション状態に直接保存することは、セッションストレージのバックエンド（SessionService）やアプリケーション全体のセキュリティ体制によっては、セキュリティリスクをもたらす可能性があります。
InMemorySessionService: テストや開発には適していますが、プロセスが終了するとデータは失われます。一時的なものであるためリスクは少ないです。- データベース/永続ストレージ: 堅牢な暗号化ライブラリ（
cryptographyなど）を使用してデータベースに保存する前にトークンデータを暗号化し、暗号化キーを安全に管理する（キー管理サービスを使用するなど）ことを強く検討してください。 - セキュアなシークレットストア: 本番環境では、機密性の高い認証情報を専用のシークレットマネージャー（Google Cloud Secret ManagerやHashiCorp Vaultなど）に保存することが最も推奨されるアプローチです。ツールは、セッション状態には短期間のアクセストークンや安全な参照のみを保存し（リフレッシュトークン自体は保存しない）、必要なときにセキュアストアから必要なシークレットを取得することができます。
ジャーニー1：認証済みツールを使用したエージェントアプリケーションの構築¶
このセクションでは、エージェントアプリケーション内で認証が必要な既存のツール（RestApiTool/OpenAPIToolset、APIHubToolset、GoogleApiToolSetなど）の使用に焦点を当てます。あなたの主な責任は、ツールの設定と、対話型認証フローのクライアント側の部分（ツールで必要な場合）の処理です。
1. 認証付きツールの設定¶
認証済みツールをエージェントに追加する際には、その必要なAuthSchemeとアプリケーションの初期AuthCredentialを提供する必要があります。
A. OpenAPIベースのツールセットの使用（OpenAPIToolset、APIHubToolsetなど）
ツールセットの初期化時にスキームと認証情報を渡します。ツールセットはそれらを生成されたすべてのツールに適用します。以下は、ADKで認証付きツールを作成するいくつかの方法です。
APIキーを必要とするツールを作成します。
from google.adk.tools.openapi_tool.auth.auth_helpers import token_to_scheme_credential
from google.adk.tools.apihub_tool.apihub_toolset import APIHubToolset
auth_scheme, auth_credential = token_to_scheme_credential(
"apikey", "query", "apikey", YOUR_API_KEY_STRING
)
sample_api_toolset = APIHubToolset(
name="sample-api-requiring-api-key",
description="APIキーで保護されたAPIを使用するツール",
apihub_resource_name="...",
auth_scheme=auth_scheme,
auth_credential=auth_credential,
)
OAuth2を必要とするツールを作成します。
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
from fastapi.openapi.models import OAuth2
from fastapi.openapi.models import OAuthFlowAuthorizationCode
from fastapi.openapi.models import OAuthFlows
from google.adk.auth import AuthCredential
from google.adk.auth import AuthCredentialTypes
from google.adk.auth import OAuth2Auth
auth_scheme = OAuth2(
flows=OAuthFlows(
authorizationCode=OAuthFlowAuthorizationCode(
authorizationUrl="https://accounts.google.com/o/oauth2/auth",
tokenUrl="https://oauth2.googleapis.com/token",
scopes={
"https://www.googleapis.com/auth/calendar": "calendar scope"
},
)
)
)
auth_credential = AuthCredential(
auth_type=AuthCredentialTypes.OAUTH2,
oauth2=OAuth2Auth(
client_id=YOUR_OAUTH_CLIENT_ID,
client_secret=YOUR_OAUTH_CLIENT_SECRET
),
)
calendar_api_toolset = OpenAPIToolset(
spec_str=google_calendar_openapi_spec_str, # openapi仕様をここに埋める
spec_str_type='yaml',
auth_scheme=auth_scheme,
auth_credential=auth_credential,
)
サービスアカウントを必要とするツールを作成します。
from google.adk.tools.openapi_tool.auth.auth_helpers import service_account_dict_to_scheme_credential
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
service_account_cred = json.loads(service_account_json_str)
auth_scheme, auth_credential = service_account_dict_to_scheme_credential(
config=service_account_cred,
scopes=["https://www.googleapis.com/auth/cloud-platform"],
)
sample_toolset = OpenAPIToolset(
spec_str=sa_openapi_spec_str, # openapi仕様をここに埋める
spec_str_type='json',
auth_scheme=auth_scheme,
auth_credential=auth_credential,
)
OpenID connectを必要とするツールを作成します。
from google.adk.auth.auth_schemes import OpenIdConnectWithConfig
from google.adk.auth.auth_credential import AuthCredential, AuthCredentialTypes, OAuth2Auth
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
auth_scheme = OpenIdConnectWithConfig(
authorization_endpoint=OAUTH2_AUTH_ENDPOINT_URL,
token_endpoint=OAUTH2_TOKEN_ENDPOINT_URL,
scopes=['openid', 'YOUR_OAUTH_SCOPES"]
)
auth_credential = AuthCredential(
auth_type=AuthCredentialTypes.OPEN_ID_CONNECT,
oauth2=OAuth2Auth(
client_id="...",
client_secret="...",
)
)
userinfo_toolset = OpenAPIToolset(
spec_str=content, # 実際の仕様を埋める
spec_str_type='yaml',
auth_scheme=auth_scheme,
auth_credential=auth_credential,
)
B. Google APIツールセットの使用（例：calendar_tool_set）
これらのツールセットには、多くの場合、専用の設定メソッドがあります。
ヒント：Google OAuthクライアントIDとシークレットの作成方法については、このガイドを参照してください：Google APIクライアントIDの取得
# 例：Googleカレンダーツールの設定
from google.adk.tools.google_api_tool import calendar_tool_set
client_id = "YOUR_GOOGLE_OAUTH_CLIENT_ID.apps.googleusercontent.com"
client_secret = "YOUR_GOOGLE_OAUTH_CLIENT_SECRET"
# このツールセットタイプ専用の設定メソッドを使用
calendar_tool_set.configure_auth(
client_id=oauth_client_id, client_secret=oauth_client_secret
)
# agent = LlmAgent(..., tools=calendar_tool_set.get_tool('calendar_tool_set'))
認証リクエストのフローのシーケンス図（ツールが認証情報を要求している場合）は以下のようになります：
2. 対話型OAuth/OIDCフローの処理（クライアント側）¶
ツールがユーザーのログイン/同意を必要とする場合（通常はOAuth 2.0またはOIDC）、ADKフレームワークは実行を一時停止し、エージェントクライアントアプリケーションに通知します。2つのケースがあります：
- エージェントクライアントアプリケーションが、同じプロセス内で直接エージェントを実行する場合（
runner.run_async経由）。例：UIバックエンド、CLIアプリ、Sparkジョブなど。 - エージェントクライアントアプリケーションが、ADKのfastapiサーバーと
/runまたは/run_sseエンドポイントを介して対話する場合。ADKのfastapiサーバーは、エージェントクライアントアプリケーションと同じサーバーまたは異なるサーバーに設定できます。
2番目のケースは1番目のケースの特殊なケースです。なぜなら、/runまたは/run_sseエンドポイントもrunner.run_asyncを呼び出すからです。唯一の違いは：
- エージェントを実行するためにPython関数を呼び出すか（1番目のケース）、サービスエンドポイントを呼び出すか（2番目のケース）。
- 結果のイベントがインメモリのオブジェクトか（1番目のケース）、HTTPレスポンス内のシリアライズされたJSON文字列か（2番目のケース）。
以下のセクションでは1番目のケースに焦点を当てており、それを2番目のケースに非常に簡単にマッピングできるはずです。必要に応じて、2番目のケースで処理すべきいくつかの違いについても説明します。
以下は、クライアントアプリケーションのステップバイステップのプロセスです：
ステップ1：エージェントの実行と認証リクエストの検出
runner.run_asyncを使用してエージェントとの対話を開始します。- yieldされたイベントを反復処理します。
- 関数呼び出しの特別な名前
adk_request_credentialを持つ特定の関数呼び出しイベントを探します。このイベントは、ユーザーの操作が必要であることを示します。ヘルパー関数を使用してこのイベントを識別し、必要な情報を抽出できます。（2番目のケースでは、ロジックは似ています。HTTPレスポンスからイベントをデシリアライズします）。
# runner = Runner(...)
# session = await session_service.create_session(...)
# content = types.Content(...) # ユーザーの初期クエリ
print("\nエージェントを実行中...")
events_async = runner.run_async(
session_id=session.id, user_id='user', new_message=content
)
auth_request_function_call_id, auth_config = None, None
async for event in events_async:
# ヘルパーを使用して特定の認証リクエストイベントを確認
if (auth_request_function_call := get_auth_request_function_call(event)):
print("--> エージェントによる認証が必要です。")
# 後で応答するために必要なIDを保存
if not (auth_request_function_call_id := auth_request_function_call.id):
raise ValueError(f'関数呼び出しから関数呼び出しIDを取得できません: {auth_request_function_call}')
# auth_uriなどを含むAuthConfigを取得
auth_config = get_auth_config(auth_request_function_call)
break # とりあえずイベントの処理を停止し、ユーザーの操作が必要
if not auth_request_function_call_id:
print("\n認証は不要か、エージェントが終了しました。")
# return # または、受信した最終応答を処理
ヘルパー関数 helpers.py：
from google.adk.events import Event
from google.adk.auth import AuthConfig # 必要な型をインポート
from google.genai import types
def get_auth_request_function_call(event: Event) -> types.FunctionCall | None:
# イベントから特別な認証リクエスト関数呼び出しを取得
if not event.content or not event.content.parts:
return None
for part in event.content.parts:
if (
part
and part.function_call
and part.function_call.name == 'adk_request_credential'
and event.long_running_tool_ids
and part.function_call.id in event.long_running_tool_ids
):
return part.function_call
return None
def get_auth_config(auth_request_function_call: types.FunctionCall) -> AuthConfig:
# 認証リクエスト関数呼び出しの引数からAuthConfigオブジェクトを抽出
if not auth_request_function_call.args or not (auth_config := auth_request_function_call.args.get('auth_config')):
raise ValueError(f'関数呼び出しから認証設定を取得できません: {auth_request_function_call}')
if not isinstance(auth_config, AuthConfig):
raise ValueError(f'認証設定 {auth_config} はAuthConfigのインスタンスではありません。')
return auth_config
ステップ2：認可のためのユーザーリダイレクト
- 前のステップで抽出した
auth_configから認可URL（auth_uri）を取得します。 - 重要なこととして、 アプリケーションの
redirect_uriをこのauth_uriにクエリパラメータとして追加します。このredirect_uriは、OAuthプロバイダーに事前登録されている必要があります（例：Google Cloud Console、Okta管理パネル）。 - ユーザーをこの完全なURLに誘導します（例：ブラウザで開く）。
# （認証が必要と検出された後の続き）
if auth_request_function_call_id and auth_config:
# AuthConfigからベースの認可URLを取得
base_auth_uri = auth_config.exchanged_auth_credential.oauth2.auth_uri
if base_auth_uri:
redirect_uri = 'http://localhost:8000/callback' # OAuthクライアントアプリの設定と一致させる必要がある
# redirect_uriを追加（本番環境ではurlencodeを使用）
auth_request_uri = base_auth_uri + f'&redirect_uri={redirect_uri}'
# ここで、エンドユーザーをこのauth_request_uriにリダイレクトするか、ブラウザで開くように依頼する必要があります
# このauth_request_uriは対応する認証プロバイダーによって提供され、エンドユーザーはログインしてアプリケーションが自分のデータにアクセスすることを承認する必要があります
# その後、認証プロバイダーはエンドユーザーを提供したredirect_uriにリダイレクトします
# 次のステップ：ユーザーからこのコールバックURLを取得する（またはWebサーバーハンドラから）
else:
print("エラー：auth_configにAuth URIが見つかりません。")
# エラーを処理
ステップ3：リダイレクトコールバックの処理（クライアント）
- アプリケーションには、ユーザーがプロバイダーでアプリケーションを承認した後にユーザーを受け取るためのメカニズム（例：
redirect_uriでのWebサーバールート）が必要です。 - プロバイダーは、
authorization_code（および潜在的にstate、scope）をクエリパラメータとしてURLに追加して、ユーザーをredirect_uriにリダイレクトします。 - この着信リクエストから完全なコールバックURLをキャプチャします。
- （このステップは、メインのエージェント実行ループの外で、Webサーバーまたは同等のコールバックハンドラ内で行われます。）
ステップ4：認証結果をADKに送り返す（クライアント）
- 完全なコールバックURL（認可コードを含む）を取得したら、クライアントステップ1で保存した
auth_request_function_call_idとauth_configオブジェクトを取得します。 - キャプチャしたコールバックURLを
exchanged_auth_credential.oauth2.auth_response_uriフィールドに設定します。また、exchanged_auth_credential.oauth2.redirect_uriに使用したリダイレクトURIが含まれていることを確認します。 types.Contentオブジェクトを作成し、types.Partにtypes.FunctionResponseを含めます。nameを"adk_request_credential"に設定します。（注：これはADKが認証を進めるための特別な名前です。他の名前は使用しないでください。）idを保存したauth_request_function_call_idに設定します。responseに、更新されたAuthConfigオブジェクトのシリアライズされた（例：.model_dump()）ものを設定します。
- この
FunctionResponseコンテンツをnew_messageとして渡し、同じセッションに対してrunner.run_asyncを再度呼び出します。
# （ユーザー操作後の続き）
# コールバックURLの取得をシミュレート（例：ユーザーのペーストやWebハンドラから）
auth_response_uri = await get_user_input(
f'完全なコールバックURLをここに貼り付けてください：\n> '
)
auth_response_uri = auth_response_uri.strip() # 入力を整形
if not auth_response_uri:
print("コールバックURLが提供されませんでした。中止します。")
return
# 受信したAuthConfigをコールバックの詳細で更新
auth_config.exchanged_auth_credential.oauth2.auth_response_uri = auth_response_uri
# トークン交換で必要になる可能性があるため、使用したredirect_uriも含める
auth_config.exchanged_auth_credential.oauth2.redirect_uri = redirect_uri
# FunctionResponse Contentオブジェクトを作成
auth_content = types.Content(
role='user', # FunctionResponseを送信する際のロールは'user'にできる
parts=[
types.Part(
function_response=types.FunctionResponse(
id=auth_request_function_call_id, # 元のリクエストへのリンク
name='adk_request_credential', # フレームワークの特別な関数名
response=auth_config.model_dump() # *更新された*AuthConfigを送り返す
)
)
],
)
# --- 実行再開 ---
print("\n認証詳細をエージェントに送り返しています...")
events_async_after_auth = runner.run_async(
session_id=session.id,
user_id='user',
new_message=auth_content, # FunctionResponseを送り返す
)
# --- 最終的なエージェント出力の処理 ---
print("\n--- 認証後のエージェント応答 ---")
async for event in events_async_after_auth:
# イベントを通常通り処理し、ツール呼び出しが成功することを期待する
print(event) # 検査のために完全なイベントを出力
ステップ5：ADKがトークン交換とツール再試行を処理し、ツールの結果を取得
- ADKは
adk_request_credentialに対するFunctionResponseを受け取ります。 - 更新された
AuthConfig内の情報（コードを含むコールバックURLなど）を使用して、プロバイダーのトークンエンドポイントでOAuthのトークン交換を実行し、アクセストークン（および場合によってはリフレッシュトークン）を取得します。 - ADKは、これらのトークンをセッション状態に設定することで内部的に利用可能にします。
- ADKは、最初に認証が不足していたために失敗した元のツール呼び出しを自動的に再試行します。
- 今回は、ツールは（
tool_context.get_auth_response()を介して）有効なトークンを見つけ、認証されたAPI呼び出しを正常に実行します。 - エージェントはツールから実際の_resultを受け取り、ユーザーへの最終的な応答を生成します。
認証応答フローのシーケンス図（エージェントクライアントが認証応答を送り返し、ADKがツール呼び出しを再試行する場合）は以下のようになります：
ジャーニー2：認証が必要なカスタムツール（FunctionTool）の構築¶
このセクションでは、新しいADKツールを作成する際に、カスタムPython関数内で認証ロジックを実装することに焦点を当てます。例としてFunctionToolを実装します。
前提条件¶
関数のシグネチャには、必ずtool_context: ToolContextを含める必要があります。ADKは、状態や認証メカニズムへのアクセスを提供するこのオブジェクトを自動的に注入します。
from google.adk.tools import FunctionTool, ToolContext
from typing import Dict
def my_authenticated_tool_function(param1: str, ..., tool_context: ToolContext) -> dict:
# ... あなたのロジック ...
pass
my_tool = FunctionTool(func=my_authenticated_tool_function)
ツール関数内の認証ロジック¶
関数内で以下のステップを実装します：
ステップ1：キャッシュされた有効な認証情報の確認
ツール関数内で、まず有効な認証情報（アクセストークン/リフレッシュトークンなど）が、このセッションの以前の実行から既に保存されているかどうかを確認します。現在のセッションの認証情報はtool_context.invocation_context.session.state（状態の辞書）に保存されている必要があります。tool_context.invocation_context.session.state.get(credential_name, None)をチェックして、既存の認証情報の有無を確認します。
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
# ツール関数内
TOKEN_CACHE_KEY = "my_tool_tokens" # 一意のキーを選択
SCOPES = ["scope1", "scope2"] # 必要なスコープを定義
creds = None
cached_token_info = tool_context.state.get(TOKEN_CACHE_KEY)
if cached_token_info:
try:
creds = Credentials.from_authorized_user_info(cached_token_info, SCOPES)
if not creds.valid and creds.expired and creds.refresh_token:
creds.refresh(Request())
tool_context.state[TOKEN_CACHE_KEY] = json.loads(creds.to_json()) # キャッシュを更新
elif not creds.valid:
creds = None # 無効、再認証が必要
tool_context.state[TOKEN_CACHE_KEY] = None
except Exception as e:
print(f"キャッシュされた認証情報の読み込み/リフレッシュ中にエラーが発生しました: {e}")
creds = None
tool_context.state[TOKEN_CACHE_KEY] = None
if creds and creds.valid:
# ステップ5に進む：認証済みAPI呼び出しを行う
pass
else:
# ステップ2に進む...
pass
ステップ2：クライアントからの認証応答の確認
- ステップ1で有効な認証情報が得られなかった場合、クライアントが対話型フローを完了したかどうかを
exchanged_credential = tool_context.get_auth_response()を呼び出して確認します。 - これは、クライアントから送り返された更新済みの
exchanged_credentialオブジェクト（auth_response_uriにコールバックURLを含む）を返します。
# ツールで設定されたauth_schemeとauth_credentialを使用。
# exchanged_credential: AuthCredential | None
exchanged_credential = tool_context.get_auth_response(AuthConfig(
auth_scheme=auth_scheme,
raw_auth_credential=auth_credential,
))
# exchanged_credentialがNoneでない場合、認証応答から交換された認証情報が既に存在する。
if exchanged_credential:
# ADKは既にアクセストークンを交換済み
access_token = exchanged_credential.oauth2.access_token
refresh_token = exchanged_credential.oauth2.refresh_token
creds = Credentials(
token=access_token,
refresh_token=refresh_token,
token_uri=auth_scheme.flows.authorizationCode.tokenUrl,
client_id=auth_credential.oauth2.client_id,
client_secret=auth_credential.oauth2.client_secret,
scopes=list(auth_scheme.flows.authorizationCode.scopes.keys()),
)
# セッション状態にトークンをキャッシュし、APIを呼び出し、ステップ5に進む
ステップ3：認証リクエストの開始
有効な認証情報（ステップ1）も認証応答（ステップ2）も見つからない場合、ツールはOAuthフローを開始する必要があります。AuthSchemeと初期のAuthCredentialを定義し、tool_context.request_credential()を呼び出します。認可が必要であることを示す応答を返します。
# ツールで設定されたauth_schemeとauth_credentialを使用。
tool_context.request_credential(AuthConfig(
auth_scheme=auth_scheme,
raw_auth_credential=auth_credential,
))
return {'pending': True, 'message': 'ユーザー認証を待っています。'}
# request_credentialを設定することで、ADKは保留中の認証イベントを検出します。実行を一時停止し、エンドユーザーにログインを求めます。
ステップ4：認可コードをトークンに交換
ADKは自動的にOAuth認可URLを生成し、それをエージェントクライアントアプリケーションに提示します。エージェントクライアントアプリケーションは、ジャーニー1で説明したのと同じ方法で、ユーザーを認可URL（redirect_uriが付加された）にリダイレクトする必要があります。ユーザーが認可URLに従ってログインフローを完了し、ADKがエージェントクライアントアプリケーションから認証コールバックURLを抽出すると、自動的に認証コードを解析し、認証トークンを生成します。次のツール呼び出し時に、ステップ2のtool_context.get_auth_responseには、後続のAPI呼び出しで使用するための有効な認証情報が含まれます。
ステップ5：取得した認証情報のキャッシュ
ADKからトークンを正常に取得した後（ステップ2）、またはトークンがまだ有効な場合（ステップ1）、新しいCredentialsオブジェクトをtool_context.stateに（シリアライズして、例：JSONとして）キャッシュキーを使用して直ちに保存します。
# ツール関数内、'creds'を取得した後（リフレッシュされたか、新しく交換されたか）
# 新しい/リフレッシュされたトークンをキャッシュ
tool_context.state[TOKEN_CACHE_KEY] = json.loads(creds.to_json())
print(f"DEBUG: トークンをキー: {TOKEN_CACHE_KEY} でキャッシュ/更新しました")
# ステップ6に進む（API呼び出し）```
**ステップ6：認証済みAPI呼び出しの実行**
* 有効な`Credentials`オブジェクト（ステップ1またはステップ4からの`creds`）を取得したら、それを使用して、適切なクライアントライブラリ（`googleapiclient`、`requests`など）を使用して保護されたAPIへの実際の呼び出しを行います。`credentials=creds`引数を渡します。
* エラーハンドリング、特に`HttpError` 401/403を含めます。これは、呼び出しの間にトークンが期限切れになったか失効したことを意味する可能性があります。そのようなエラーが発生した場合は、キャッシュされたトークンをクリアし（`tool_context.state.pop(...)`）、再認証を強制するために`auth_required`ステータスを再度返すことを検討してください。
```py
# ツール関数内、有効な'creds'オブジェクトを使用
# 続行する前にcredsが有効であることを確認
if not creds or not creds.valid:
return {"status": "error", "error_message": "有効な認証情報なしでは続行できません。"}
try:
service = build("calendar", "v3", credentials=creds) # 例
api_result = service.events().list(...).execute()
# ステップ7に進む
except Exception as e:
# APIエラーを処理（例：401/403をチェックし、キャッシュをクリアして再認証をリクエストするかもしれない）
print(f"エラー：API呼び出しが失敗しました: {e}")
return {"status": "error", "error_message": f"API呼び出しが失敗しました: {e}"}
ステップ7：ツール結果の返却
- API呼び出しが成功した後、結果をLLMにとって有用な辞書形式に処理します。
- 重要なこととして、 データと共にステータスを含めます。
# ツール関数内、API呼び出しが成功した後
processed_result = [...] # LLM用にapi_resultを処理
return {"status": "success", "data": processed_result}
完全なコード
tools_and_agent.py
import os
from google.adk.auth.auth_schemes import OpenIdConnectWithConfig
from google.adk.auth.auth_credential import AuthCredential, AuthCredentialTypes, OAuth2Auth
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
from google.adk.agents.llm_agent import LlmAgent
# --- Authentication Configuration ---
# This section configures how the agent will handle authentication using OpenID Connect (OIDC),
# often layered on top of OAuth 2.0.
# Define the Authentication Scheme using OpenID Connect.
# This object tells the ADK *how* to perform the OIDC/OAuth2 flow.
# It requires details specific to your Identity Provider (IDP), like Google OAuth, Okta, Auth0, etc.
# Note: Replace the example Okta URLs and credentials with your actual IDP details.
# All following fields are required, and available from your IDP.
auth_scheme = OpenIdConnectWithConfig(
# The URL of the IDP's authorization endpoint where the user is redirected to log in.
authorization_endpoint="https://your-endpoint.okta.com/oauth2/v1/authorize",
# The URL of the IDP's token endpoint where the authorization code is exchanged for tokens.
token_endpoint="https://your-token-endpoint.okta.com/oauth2/v1/token",
# The scopes (permissions) your application requests from the IDP.
# 'openid' is standard for OIDC. 'profile' and 'email' request user profile info.
scopes=['openid', 'profile', "email"]
)
# Define the Authentication Credentials for your specific application.
# This object holds the client identifier and secret that your application uses
# to identify itself to the IDP during the OAuth2 flow.
# !! SECURITY WARNING: Avoid hardcoding secrets in production code. !!
# !! Use environment variables or a secret management system instead. !!
auth_credential = AuthCredential(
auth_type=AuthCredentialTypes.OPEN_ID_CONNECT,
oauth2=OAuth2Auth(
client_id="CLIENT_ID",
client_secret="CIENT_SECRET",
)
)
# --- Toolset Configuration from OpenAPI Specification ---
# This section defines a sample set of tools the agent can use, configured with Authentication
# from steps above.
# This sample set of tools use endpoints protected by Okta and requires an OpenID Connect flow
# to acquire end user credentials.
with open(os.path.join(os.path.dirname(__file__), 'spec.yaml'), 'r') as f:
spec_content = f.read()
userinfo_toolset = OpenAPIToolset(
spec_str=spec_content,
spec_str_type='yaml',
# ** Crucially, associate the authentication scheme and credentials with these tools. **
# This tells the ADK that the tools require the defined OIDC/OAuth2 flow.
auth_scheme=auth_scheme,
auth_credential=auth_credential,
)
# --- Agent Configuration ---
# Configure and create the main LLM Agent.
root_agent = LlmAgent(
model='gemini-2.0-flash',
name='enterprise_assistant',
instruction='Help user integrate with multiple enterprise systems, including retrieving user information which may require authentication.',
tools=userinfo_toolset.get_tools(),
)
# --- Ready for Use ---
# The `root_agent` is now configured with tools protected by OIDC/OAuth2 authentication.
# When the agent attempts to use one of these tools, the ADK framework will automatically
# trigger the authentication flow defined by `auth_scheme` and `auth_credential`
# if valid credentials are not already available in the session.
# The subsequent interaction flow would guide the user through the login process and handle
# token exchanging, and automatically attach the exchanged token to the endpoint defined in
# the tool.
agent_cli.py
import asyncio
from dotenv import load_dotenv
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from .helpers import is_pending_auth_event, get_function_call_id, get_function_call_auth_config, get_user_input
from .tools_and_agent import root_agent
load_dotenv()
agent = root_agent
async def async_main():
"""
Main asynchronous function orchestrating the agent interaction and authentication flow.
"""
# --- Step 1: Service Initialization ---
# Use in-memory services for session and artifact storage (suitable for demos/testing).
session_service = InMemorySessionService()
artifacts_service = InMemoryArtifactService()
# Create a new user session to maintain conversation state.
session = session_service.create_session(
state={}, # Optional state dictionary for session-specific data
app_name='my_app', # Application identifier
user_id='user' # User identifier
)
# --- Step 2: Initial User Query ---
# Define the user's initial request.
query = 'Show me my user info'
print(f"user: {query}")
# Format the query into the Content structure expected by the ADK Runner.
content = types.Content(role='user', parts=[types.Part(text=query)])
# Initialize the ADK Runner
runner = Runner(
app_name='my_app',
agent=agent,
artifact_service=artifacts_service,
session_service=session_service,
)
# --- Step 3: Send Query and Handle Potential Auth Request ---
print("\nRunning agent with initial query...")
events_async = runner.run_async(
session_id=session.id, user_id='user', new_message=content
)
# Variables to store details if an authentication request occurs.
auth_request_event_id, auth_config = None, None
# Iterate through the events generated by the first run.
async for event in events_async:
# Check if this event is the specific 'adk_request_credential' function call.
if is_pending_auth_event(event):
print("--> Authentication required by agent.")
auth_request_event_id = get_function_call_id(event)
auth_config = get_function_call_auth_config(event)
# Once the auth request is found and processed, exit this loop.
# We need to pause execution here to get user input for authentication.
break
# If no authentication request was detected after processing all events, exit.
if not auth_request_event_id or not auth_config:
print("\nAuthentication not required for this query or processing finished.")
return # Exit the main function
# --- Step 4: Manual Authentication Step (Simulated OAuth 2.0 Flow) ---
# This section simulates the user interaction part of an OAuth 2.0 flow.
# In a real web application, this would involve browser redirects.
# Define the Redirect URI. This *must* match one of the URIs registered
# with the OAuth provider for your application. The provider sends the user
# back here after they approve the request.
redirect_uri = 'http://localhost:8000/dev-ui' # Example for local development
# Construct the Authorization URL that the user must visit.
# This typically includes the provider's authorization endpoint URL,
# client ID, requested scopes, response type (e.g., 'code'), and the redirect URI.
# Here, we retrieve the base authorization URI from the AuthConfig provided by ADK
# and append the redirect_uri.
# NOTE: A robust implementation would use urlencode and potentially add state, scope, etc.
auth_request_uri = (
auth_config.exchanged_auth_credential.oauth2.auth_uri
+ f'&redirect_uri={redirect_uri}' # Simple concatenation; ensure correct query param format
)
print("\n--- User Action Required ---")
# Prompt the user to visit the authorization URL, log in, grant permissions,
# and then paste the *full* URL they are redirected back to (which contains the auth code).
auth_response_uri = await get_user_input(
f'1. Please open this URL in your browser to log in:\n {auth_request_uri}\n\n'
f'2. After successful login and authorization, your browser will be redirected.\n'
f' Copy the *entire* URL from the browser\'s address bar.\n\n'
f'3. Paste the copied URL here and press Enter:\n\n> '
)
# --- Step 5: Prepare Authentication Response for the Agent ---
# Update the AuthConfig object with the information gathered from the user.
# The ADK framework needs the full response URI (containing the code)
# and the original redirect URI to complete the OAuth token exchange process internally.
auth_config.exchanged_auth_credential.oauth2.auth_response_uri = auth_response_uri
auth_config.exchanged_auth_credential.oauth2.redirect_uri = redirect_uri
# Construct a FunctionResponse Content object to send back to the agent/runner.
# This response explicitly targets the 'adk_request_credential' function call
# identified earlier by its ID.
auth_content = types.Content(
role='user',
parts=[
types.Part(
function_response=types.FunctionResponse(
# Crucially, link this response to the original request using the saved ID.
id=auth_request_event_id,
# The special name of the function call we are responding to.
name='adk_request_credential',
# The payload containing all necessary authentication details.
response=auth_config.model_dump(),
)
)
],
)
# --- Step 6: Resume Execution with Authentication ---
print("\nSubmitting authentication details back to the agent...")
# Run the agent again, this time providing the `auth_content` (FunctionResponse).
# The ADK Runner intercepts this, processes the 'adk_request_credential' response
# (performs token exchange, stores credentials), and then allows the agent
# to retry the original tool call that required authentication, now succeeding with
# a valid access token embedded.
events_async = runner.run_async(
session_id=session.id,
user_id='user',
new_message=auth_content, # Provide the prepared auth response
)
# Process and print the final events from the agent after authentication is complete.
# This stream now contain the actual result from the tool (e.g., the user info).
print("\n--- Agent Response after Authentication ---")
async for event in events_async:
print(event)
if __name__ == '__main__':
asyncio.run(async_main())
helpers.py
from google.adk.auth import AuthConfig
from google.adk.events import Event
import asyncio
# --- Helper Functions ---
async def get_user_input(prompt: str) -> str:
"""
Asynchronously prompts the user for input in the console.
Uses asyncio's event loop and run_in_executor to avoid blocking the main
asynchronous execution thread while waiting for synchronous `input()`.
Args:
prompt: The message to display to the user.
Returns:
The string entered by the user.
"""
loop = asyncio.get_event_loop()
# Run the blocking `input()` function in a separate thread managed by the executor.
return await loop.run_in_executor(None, input, prompt)
def is_pending_auth_event(event: Event) -> bool:
"""
Checks if an ADK Event represents a request for user authentication credentials.
The ADK framework emits a specific function call ('adk_request_credential')
when a tool requires authentication that hasn't been previously satisfied.
Args:
event: The ADK Event object to inspect.
Returns:
True if the event is an 'adk_request_credential' function call, False otherwise.
"""
# Safely checks nested attributes to avoid errors if event structure is incomplete.
return (
event.content
and event.content.parts
and event.content.parts[0] # Assuming the function call is in the first part
and event.content.parts[0].function_call
# The specific function name indicating an auth request from the ADK framework.
and event.content.parts[0].function_call.name == 'adk_request_credential'
)
def get_function_call_id(event: Event) -> str:
"""
Extracts the unique ID of the function call from an ADK Event.
This ID is crucial for correlating a function *response* back to the specific
function *call* that the agent initiated to request for auth credentials.
Args:
event: The ADK Event object containing the function call.
Returns:
The unique identifier string of the function call.
Raises:
ValueError: If the function call ID cannot be found in the event structure.
(Corrected typo from `contents` to `content` below)
"""
# Navigate through the event structure to find the function call ID.
if (
event
and event.content
and event.content.parts
and event.content.parts[0] # Use content, not contents
and event.content.parts[0].function_call
and event.content.parts[0].function_call.id
):
return event.content.parts[0].function_call.id
# If the ID is missing, raise an error indicating an unexpected event format.
raise ValueError(f'Cannot get function call id from event {event}')
def get_function_call_auth_config(event: Event) -> AuthConfig:
"""
Extracts the authentication configuration details from an 'adk_request_credential' event.
Client should use this AuthConfig to necessary authentication details (like OAuth codes and state)
and sent it back to the ADK to continue OAuth token exchanging.
Args:
event: The ADK Event object containing the 'adk_request_credential' call.
Returns:
An AuthConfig object populated with details from the function call arguments.
Raises:
ValueError: If the 'auth_config' argument cannot be found in the event.
(Corrected typo from `contents` to `content` below)
"""
if (
event
and event.content
and event.content.parts
and event.content.parts[0] # Use content, not contents
and event.content.parts[0].function_call
and event.content.parts[0].function_call.args
and event.content.parts[0].function_call.args.get('auth_config')
):
# Reconstruct the AuthConfig object using the dictionary provided in the arguments.
# The ** operator unpacks the dictionary into keyword arguments for the constructor.
return AuthConfig(
**event.content.parts[0].function_call.args.get('auth_config')
)
raise ValueError(f'Cannot get auth config from event {event}')