🤖
Agent Development Kit 1.7.0 で追加された Plugin 機能

2025/07/17に公開
Vertex AI
LLM
AIエージェント
Agent Development Kit
tech
こんにちはサントリーこと大橋です。

先ほど Agent Development Kit(ADK) 1.7.0がリリースされました。
https://github.com/google/adk-python/releases/tag/v1.7.0
主な追加機能は
Local Eval(評価)の実行結果の永続化
Plugin機能の追加
(リリースノートには無いけど) YAML経由でのAgentの作成
これはまだWIPな機能です。

です。

今回は影響度が大きい 「Plugin機能の追加」 について、検証していきたいと思います。

 ADKのPlugin機能
 今までのADKの課題これまでのADKには、各処理の実行前後に任意の処理を差し込める callbacks という機能がありました。
https://google.github.io/adk-docs/callbacks/
from google.adk.agents import LlmAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmResponse, LlmRequest
from typing import Optional

# --- Define your callback function ---
def my_before_model_logic(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    print(f"Callback running before model call for agent: {callback_context.agent_name}")
    # ... your custom logic here ...
    return None # Allow the model call to proceed

# --- Register it during Agent creation ---
my_agent = LlmAgent(
    name="MyCallbackAgent",
    model="gemini-2.0-flash", # Or your desired model
    instruction="Be helpful.",
    # Other agent parameters...
    before_model_callback=my_before_model_logic # Pass the function here
)
※ ADKのcallback
しかし、この機能はAgentごとに設定する必要があり、例えば処理前後にログを差し込みたい場合は、すべてのAgentに対して、このcallbackを設定する必要がありました。

 ADKのPlugin機能ADK 1.7.0で追加された Plugin は上記課題を解決できる機能で、Runnerに対してPluginを設定することで、あらかじめ定義されたポイントに共通の処理を差し込める機能です。
runner = Runner(
      app_name="my-app",
      agent=root_agent,
      artifact_service=artifact_service,
      session_service=session_service,
      memory_service=memory_service,
      plugins=[
        MySamplePlugin(),
        LoggingPlugin(),
      ],
  )
※ ADKのPluginの設定方法
callbackが特定のAgentに設定するのに対し、Pluginはアプリケーション全体に横断的な処理を差し込めるようになっています。
例えば、Pluginは以下のような用途に活用できます。
ログ記録・追跡: エージェントやツールの活動を詳細に記録し、デバッグやパフォーマンス分析に利用する。
ポリシー適用: 特定のツール使用前にユーザー権限を確認するなど、セキュリティ上のガードレールを実装する。
監視・メトリクス収集: トークン使用量や実行時間などの指標を収集し、監視システムに送信する。
キャッシュ: LLMやツールを呼び出す前に、同じリクエストが過去にあったかを確認し、あればキャッシュした結果を返すことで、高コストな処理をスキップする。

*リクエスト/レスポンスの変更: LLMへのプロンプトに動的に情報を追加したり、ツールの出力を標準化したりする。
詳しい仕様は、以下の公式ドキュメントを参照してください。
https://github.com/google/adk-python/blob/main/contributing/samples/plugin_basic/README.md
なお残念ながら現状このPlugin機能はadk web経由では利用できませんのでご注意ください。

 Pluginの実装/仕様方法
 Pluginの作成ではPluginを作成してみましょう。
Pluginを作成するには google.adk.plugins.base_plugin.BasePluginを継承したクラスを作成し、処理を差し込みたいポイントに対応するメソッドをオーバーライドします。
差し込める対象の処理は以下です。
on_user_message_callback
before_run_callback
after_run_callback
on_event_callback
before_agent_callback
after_agent_callback
before_tool_callback
after_tool_callback
before_model_callback
after_model_callback
基本的なフックポイントはcallbacksと同様ですが、いくつか追加・変更点があるようです。
今回は、すべてのポイントでログを差し込むLoggerPluginを作成してみます。
main.py
import logging
from typing import Optional, Any

from google.adk.agents import Agent, InvocationContext, BaseAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.events import Event
from google.adk.models import LlmRequest, LlmResponse
from google.adk.runners import InMemoryRunner
from google.adk.plugins import BasePlugin
from google.adk.tools import BaseTool, ToolContext
from google.genai import types

class LoggerPlugin(BasePlugin):

    def __init__(self):
        super().__init__(name="LoggerPlugin")

    async def on_user_message_callback(
            self,
            *,
            invocation_context: InvocationContext,
            user_message: types.Content,
    ) -> Optional[types.Content]:
        print(f"on_user_message_callback")

    async def before_run_callback(
            self, *, invocation_context: InvocationContext
    ) -> Optional[types.Content]:
        print("before_run_callback")

    async def on_event_callback(
            self, *, invocation_context: InvocationContext, event: Event
    ) -> Optional[Event]:
        print(f"on_event_callback")

    async def after_run_callback(
            self, *, invocation_context: InvocationContext
    ) -> Optional[None]:
        print(f"after_run_callback")

    async def before_agent_callback(
            self, *, agent: BaseAgent, callback_context: CallbackContext
    ) -> Optional[types.Content]:
        print(f"before_agent_callback agent: {agent.name}")

    async def after_agent_callback(
            self, *, agent: BaseAgent, callback_context: CallbackContext
    ) -> Optional[types.Content]:
        print(f"after_agent_callback agent: {agent.name}")

    async def before_model_callback(
            self, *, callback_context: CallbackContext, llm_request: LlmRequest
    ) -> Optional[LlmResponse]:
        print(f"before_model_callback")

    async def after_model_callback(
            self, *, callback_context: CallbackContext, llm_response: LlmResponse
    ) -> Optional[LlmResponse]:
        print(f"after_model_callback")

    async def before_tool_callback(
            self,
            *,
            tool: BaseTool,
            tool_args: dict[str, Any],
            tool_context: ToolContext,
    ) -> Optional[dict]:
        print(f"before_tool_callback tool:{tool.name}")

    async def after_tool_callback(
            self,
            *,
            tool: BaseTool,
            tool_args: dict[str, Any],
            tool_context: ToolContext,
            result: dict,
    ) -> Optional[dict]:
        print(f"after_tool_callback tool:{tool.name}")

※ Pluginの実装
実装は非常にシンプルですね。

それでは実際にRunner経由で実行してみましょう。
main.py
import datetime
import logging
from typing import Optional, Any

from google.adk.agents import Agent, InvocationContext, BaseAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.events import Event
from google.adk.models import LlmRequest, LlmResponse
from google.adk.runners import InMemoryRunner
from google.adk.plugins import BasePlugin
from google.adk.tools import BaseTool, ToolContext
from google.genai import types
import dotenv

dotenv.load_dotenv()


logger = logging.getLogger()


def now_tool():
    """
    現在時刻を返却します。
    Returns: 現在時刻
    """
    return datetime.datetime.now()


root_agent = Agent(
    model='gemini-2.5-flash',
    name='root_agent',
    description='A helpful assistant for user questions.',
    instruction='Answer user questions to the best of your knowledge, when you are asked about now time, use `now_tool`',
    tools=[now_tool]
)


async def call_agent_async(query: str, runner: InMemoryRunner, user_id, session_id):
    """Sends a query to the agent and prints the final response."""
    print(f"\n>>> User Query: {query}")

    content = types.Content(role='user', parts=[types.Part(text=query)])
    final_response_text = "Agent did not produce a final response."
    async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
        if event.is_final_response():
            if event.content and event.content.parts:
                # Assuming text response in the first part
                final_response_text = event.content.parts[0].text
            elif event.actions and event.actions.escalate:  # Handle potential errors/escalations
                final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
            # Add more checks here if needed (e.g., specific error codes)
            break  # Stop processing events once the final response is found
    print(f"<<< Agent Response: {final_response_text}")


async def main():
    runner = InMemoryRunner(
        app_name="test",
        agent=root_agent,
        plugins=[
            LoggerPlugin(),
        ]
    )

    await runner.session_service.create_session(app_name="test", user_id="test", session_id="test")

    await call_agent_async("こんにちは", runner, "test", "test")
    await call_agent_async("今何時", runner, "test", "test")
    await call_agent_async("何ができる？", runner, "test", "test")


import asyncio

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"An error occurred: {e}")

※ 実行用スクリプト(main.py)
単純なagentと現在時刻を取得するツールを用意して実行してみます。
 uv run plugin_sample/agent.py

>>> User Query: こんにちは
on_user_message_callback
before_run_callback
before_agent_callback agent: root_agent
before_model_callback
before_model_callback
on_event_callback
<<< Agent Response: こんにちは！何かお手伝いできることはありますか？

>>> User Query: 今何時
on_user_message_callback
before_run_callback
before_agent_callback agent: root_agent
before_model_callback
Warning: there are non-text parts in the response: ['function_call'], returning concatenated text result from text parts. Check the full candidates.content.parts accessor to get the full model response.
before_model_callback
on_event_callback
before_tool_callback tool:now_tool
after_tool_callback tool:now_tool
on_event_callback
before_model_callback
before_model_callback
on_event_callback
<<< Agent Response: 現在時刻は2025年7月17日15時32分です。

>>> User Query: 何ができる？
on_user_message_callback
before_run_callback
before_agent_callback agent: root_agent
before_model_callback
before_model_callback
on_event_callback
<<< Agent Response: 私は質問に答えたり、情報を提供したりすることができます。また、現在時刻をお伝えすることもできます。
ログが表示されましたね。

callbackを作成したことがある方にとっては馴染み深く、それほど難しくなく実装できると思います。

 まとめ今回は今日リリースされた ADK 1.7.0 の新機能 Plugin について説明しました。

トレーサビリティの向上や認証処理の実装など、様々な観点で活用できるでしょう。
機会があれば、YAML経由でのAgent作成についても解説したいと思います。

 お知らせ/宣伝先日、ADKユーザー向けのイベント「ADK UG #0」が開催され、ADK開発者が集う日本語のDiscordコミュニティが誕生しました。ADKに関する情報交換や議論に興味がある方は、ぜひご参加ください！
https://discord.gg/BKpGRzjtqZ
また、ADKの最新のコミットログやリリースノートを分かりやすく解説するPodcastを、月・水・金に配信しています。ADKの動向を追いかけたい方は、ぜひ聴いてみてください。
https://www.youtube.com/playlist?list=PL0Zc2RFDZsM_MkHOzWNJpaT4EH5fQxA8n
ADKのPlugin機能

今までのADKの課題

ADKのPlugin機能

Pluginの実装/仕様方法

Pluginの作成

まとめ

お知らせ/宣伝

Discussion
