LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@hina02limited

ADKでGoogle Calendar等のOahtu2認証が必要なツールセットを活用するエージェントを作成してみました。

Posted at

目次

1_ツールセット
2_ツールの認証プロセス
3_アクセストークンを用いた認証
4_まとめ

Googleの「Agent Development Kit (ADK)」というAIエージェント開発フレームワークを使って、Google Calendar等のOahtu2認証が必要なツールセットを活用するエージェントを作成してみました。

基本的に公式ドキュメントの内容に沿って進めましたが、そのまま適用するとエラーになったり、アクセストークンの再利用がうまくいかない問題に直面しました。
そこで、ソースコードを調べて解決策を見つけたので、その過程を記事にまとめました。どなたかの参考になれば幸いです。

1_ツールセット

ツールセットの呼び出し

ADKがデフォルトで用意しているGoogle APIのツールセットは、以下のコードで呼び出すことができます。

from google.adk.tools.google_api_tool import GoogleApiToolset
from google.adk.tools.google_api_tool import BigQueryToolset
from google.adk.tools.google_api_tool import CalendarToolset
from google.adk.tools.google_api_tool import DocsToolset
from google.adk.tools.google_api_tool import GmailToolset
from google.adk.tools.google_api_tool import SheetsToolset
from google.adk.tools.google_api_tool import SlidesToolset
from google.adk.tools.google_api_tool import YoutubeToolset

一見すると、ここに書かれているツールセットしか使えないように見えますが、例えばCalendarToolsetクラスのソースコードを見ると、実はGoogleApiToolsetを継承し、初期化時にapi_name = "calendar"とapi_version = "v3"を渡しているだけだと分かります。

class CalendarToolset(GoogleApiToolset):
  """Auto-generated Calendar toolset based on Google Calendar API v3 spec exposed by Google API discovery API"""

  def __init__(
      self,
      client_id: str = None,
      client_secret: str = None,
      tool_filter: Optional[Union[ToolPredicate, List[str]]] = None,
  ):
    super().__init__("calendar", "v3", client_id, client_secret, tool_filter)

そのため、GoogleApiToolsetに渡すapi_nameとapi_versionを変えるだけで、Google Drive、Task、Chat、Forms、Looker等々、多数のGoogle APIに関するツールセットを利用できるようになります。

これはとんでもないｗ

正直なところ、このgoogle_api_toolだけで、ADKを採用する価値はあるんじゃないかと思います。

ツールセットのフィルタリング

各ツールセットに含まれるツールの情報は、tool_filterにリストやCallableな関数を渡すことで、フィルタリングすることができます。

toolset = GoogleApiToolset(
    api_name="calendar",
    api_version="v3",
    tool_filter=lambda tool, readonly_context: tool.name.endswith("get"),
)
tools = await toolset.get_tools() # ツールのリストを取得

この例は、tool.nameがcalendar_calendars_get等のgetで終わるものに限定したケースです。
このようにすることで、読み取り専用ツールしか持たないエージェントを作成し、エージェントが誤って削除ツールを使用したりしてしまうリスクを回避することができます。

2_ツールの認証プロセス

OAuth 2.0 クライアント IDの作成

事前に、デスクトップ用のOAuth 2.0 クライアント IDを作成して、client_idとclient_secretを取得しておきます。

ヘルパー関数

公式ドキュメントのコードを一部変更して使用しました。（そのままだとエラーになったため）

  • get_auth_request_function_call
    • 条件式を修正
  • get_auth_config
    • パラメータ名を'AuthConfig'に修正
    • AuthConfigクラスに変換して返すように変更
      （後続でAuthConfig.exchanged_auth_credentialを使う箇所があるため）
from google.adk.events import Event
from google.genai import types
from google.adk.auth import AuthConfig

def get_auth_request_function_call(event: Event) -> types.FunctionCall:
    """"adk_request_credential"を呼び出しているイベントを検出する関数"""
    if not (event.content and event.content.parts):  # 変更箇所
        return
    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


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("authConfig")  # 変更箇所
    ):
        raise ValueError(f"Cannot get auth config from function call: {auth_request_function_call}")
    return AuthConfig.model_validate(auth_config)  # 変更箇所


async def get_user_input(prompt: str) -> str:
    """ターミナルでユーザーの入力を待つ関数"""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, input, prompt)

メイン関数

公式ドキュメントのコードを繋げて、エージェントの作成、RunnerとSessionの作成等を捕捉しました。

これを実行すれば、ターミナルにURLが表示されるので、認証プロセスを完了し、URLをターミナルに貼り付けてください。

認証のため、次のURLをブラウザで開いてください: <URL>
認証後の完全なコールバックURLを貼り付けてください:
>

すると、以下のようなレスポンスが返ってきて、実際にGoogle Calendarに予定が追加されます。

Agent Response:  OK。CalendarID = '***'に、JSTで2025/6/5の18:00にディナーの予定を追加しました

from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types
from google.adk.tools.google_api_tool import CalendarToolset

client_id = # 作成したOAuth2.0クライアントのclient_id
client_secret = # 作成したOAuth2.0クライアントのclient_secret

async def main():
    # ツールセットの呼び出し
    calendar_tool_set = CalendarToolset()
    calendar_tool_set.configure_auth(client_id=client_id, client_secret=client_secret)
    tools = await calendar_tool_set.get_tools()

    # エージェントの作成
    agent = LlmAgent(
        name="calendar_agent",
        model="gemini-2.0-flash",
        description=("Agent to answer questions about the google calendar."),
        instruction=(
            """You are a helpful agent who can answer user questions about the user calendar.
                must use the calendar tool to answer the user question.
                must use function call."""
        ),
        tools=tools,
    )

    # Runnerの作成とSessionの作成
    APP_NAME = "auth_test"
    USER_ID = "user_1"
    session_service = InMemorySessionService()
    runner = Runner(
        agent=agent,
        app_name=APP_NAME,
        session_service=session_service,
    )
    print(f"Runner created for agent '{runner.agent.name}'.")

    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID)
    print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{session.id}'")

    # クエリを渡して、Runnerを実行。
    query = "JSTで2025/6/5の18:00にディナーの予定を追加して CalendarID = ***"
    content = types.Content(role="user", parts=[types.Part(text=query)])
    events_async = runner.run_async(session_id=session.id, user_id=USER_ID, new_message=content)

    # ------------メインプロセス------------

    # Step1 イベントループで、"adk_request_credential"を検出
    auth_request_function_call_id = None
    auth_config = None
    async for event in events_async:
        if auth_request_function_call := get_auth_request_function_call(event):
            print("--> Authentication required by agent.")
            if not (auth_request_function_call_id := auth_request_function_call.id):
                raise ValueError(
                    f"Cannot get function call id from function call: {auth_request_function_call}"
                )
            auth_config = get_auth_config(auth_request_function_call)
            print(f"Auth config: {auth_config}")
            break

        if not auth_request_function_call_id:
            print("\nAuth not required or agent finished.")

    # Step 2　認証のためにユーザーをリダイレクトする
    if auth_request_function_call_id and auth_config:
        redirect_uri = "http://localhost:8000"
        base_auth_uri = auth_config.exchanged_auth_credential.oauth2.auth_uri
        if base_auth_uri:
            auth_request_uri = base_auth_uri + f"&redirect_uri={redirect_uri}"
            print(f"認証のため、次のURLをブラウザで開いてください: {auth_request_uri}")
        else:
            print("ERROR: Auth URI not found in auth_config.")

        # Step 3
        auth_response_uri = await get_user_input(
            "認証後の完全なコールバックURLを貼り付けてください:\n> "
        )
        auth_response_uri = auth_response_uri.strip()

        if not auth_response_uri:
            print("Callback URL not provided. Aborting.")
            return

        # コールバックURLから認証コードを抽出してaccess_tokenに交換
        token_data = exchange_code_for_tokens(
            auth_response_uri, oauth_client_id, oauth_client_secret, redirect_uri
        )
        print(f"Token data: {token_data}")

        # AuthConfigをコールバックに基づいて更新
        auth_config.exchanged_auth_credential.oauth2.auth_response_uri = auth_response_uri
        auth_config.exchanged_auth_credential.oauth2.redirect_uri = redirect_uri
        auth_config.exchanged_auth_credential.oauth2.access_token = token_data.get("access_token")

        # 4. 認証結果をエージェントに送る
        auth_content = types.Content(
            role="user",
            parts=[
                types.Part(
                    function_response=types.FunctionResponse(
                        id=auth_request_function_call_id,
                        name="adk_request_credential",
                        response=auth_config.model_dump(),
                    ),
                ),
            ],
        )
        print(f"Auth content: {auth_content}")

        # --- Resume Execution ---
        print("\nSubmitting authentication details back to the agent...")
        sessions = await session_service.list_sessions(app_name=APP_NAME, user_id=USER_ID)
        print(f"Sessions: {sessions}")
        events_async_after_auth = runner.run_async(
            session_id=session.id,
            user_id=USER_ID,
            new_message=auth_content,  # Send the FunctionResponse back
        )

        # --- Process Final Agent Output ---
        print("\n--- Agent Response after Authentication ---")
        async for event in events_async_after_auth:
            print(event.content.parts[0].text)

    # 5. 最終的なレスポンスを確認
    if event.is_final_response():
        final_response = event.content.parts[0].text
        print("Agent Response: ", final_response)

3_アクセストークンを用いた認証

上記の公式ドキュメントの例では、単発の認証プロセスの実行のみですが、実際にアプリケーションを構築するとなると、アクセストークンを用いて認証プロセスを省略したいところです。

先ほどのコードを実行した際に、ターミナルに表示されるAuth contentの中に、access_tokenが含まれています。これをツールセットの呼び出しのところで、渡してやることで、アクセストークンを使って認証することができます。

以下のコードの箇所を入れ替えて実行することで、アクセストークンを用いた認証ができました。

access_token = # アクセストークンを貼り付けてください。

async def main():
    # ツールセットの呼び出し
    calendar_tool_set = CalendarToolset()
    calendar_tool_set.configure_auth(client_id=oauth_client_id, client_secret=oauth_client_secret)
    if access_token:
        # OpenAPIToolset形式に変換して、認証情報を更新
        auth_scheme, auth_credential = token_to_scheme_credential(
            token_type="oauth2Token",
            location="header",
            name="Authorization",
            credential_value=access_token,
        )
        calendar_tool_set = calendar_tool_set._openapi_toolset
        calendar_tool_set._configure_auth_all(auth_scheme, auth_credential)

    tools = await calendar_tool_set.get_tools()

このコードでは、token_to_scheme_credentialを使って、アクセストークンからAuthSchemeとAuthCredentialを生成しています。

CalendarToolset(GoogleAPIToolset)は、そのままの型ではこれらの値を渡すことができないため、._openapi_toolsetでOpenAPIToolset形式に変換しています。

ここからget_toolsを使うことで、認証情報を持ったツールが用意できます。

4_まとめ

以上、ADKで一番使いたかった、OAuth2認証が必要なツールを活用したエージェントの作成について解説しました。

アクセストークンについては、現状ではやや裏技的な方法で認証をクリアしましたが、正規の方法でアクセストークン行けるようになったよってなったら、コメント等で教えていただけるとありがたいです。

冒頭でも述べた通りADKは豊富なツールセットを持つという点だけで十分に魅力的です。
エージェントフレームワークとしても、LangGraph、CrewAI、PydanticAI等の良いところをちゃんと押さえた正当進化系のようで、今個人的に一押しのエージェントフレームワークです。

A2Aと合わせて、どんどん使っていきましょう。

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?