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

## 目次
[1_ツールセット](#1_ツールセット)
[2_ツールの認証プロセス](#2_ツールの認証プロセス)
[3_アクセストークンを用いた認証](#3_アクセストークンを用いた認証)
[4_まとめ](#4_まとめ)



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

https://google.github.io/adk-docs/tools/authentication/#1-configuring-tools-with-authentication

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

# 1_ツールセット
### ツールセットの呼び出し
ADKがデフォルトで用意しているGoogle APIのツールセットは、以下のコードで呼び出すことができます。
```python
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"`を渡しているだけだと分かります。
```python
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な関数を渡すことで、フィルタリングすることができます。

```python
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を取得しておきます。

https://console.developers.google.com/?hl=ja

https://support.google.com/workspacemigrate/answer/9222992?hl=ja

### ヘルパー関数
公式ドキュメントのコードを一部変更して使用しました。（そのままだとエラーになったため）
- get_auth_request_function_call
    - 条件式を修正
- get_auth_config
    - パラメータ名を'AuthConfig'に修正
    - AuthConfigクラスに変換して返すように変更
（後続でAuthConfig.exchanged_auth_credentialを使う箇所があるため）

```python
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にディナーの予定を追加しました
```


```python
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**が含まれています。これをツールセットの呼び出しのところで、渡してやることで、アクセストークンを使って認証することができます。

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


```python
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と合わせて、どんどん使っていきましょう。