第三方工具¶

Agent Development Kit (ADK) 設計上高度可擴充，讓你能無縫整合來自其他 AI Agent 框架（如 CrewAI 和 LangChain）的工具。這種互通性非常關鍵，因為它能加速開發流程，並讓你重複利用現有工具。

1. 使用 LangChain 工具¶

Agent Development Kit (ADK) 提供 LangchainTool 包裝器，讓你可以將 LangChain 生態系統中的工具整合到你的 agent 中。

範例：使用 LangChain 的 Tavily 工具進行網路搜尋¶

Tavily 提供一個搜尋 API，能根據即時搜尋結果回傳答案，適合 AI agent 等應用程式使用。

請依照 ADK 安裝與設定指南操作。
安裝相依套件：請確認你已安裝所需的 LangChain 套件。例如，若要使用 Tavily 搜尋工具，請安裝其專屬的相依套件：
```
pip install langchain_community tavily-python
```
取得 Tavily API KEY，並將其匯出為環境變數。
```
export TAVILY_API_KEY=<REPLACE_WITH_API_KEY>
```
匯入： 從 Agent Development Kit (ADK) 匯入 LangchainTool 包裝器，以及你想使用的特定 LangChain 工具（例如 TavilySearchResults）。
```
from google.adk.tools.langchain_tool import LangchainTool
from langchain_community.tools import TavilySearchResults
```

實例化與包裝： 建立你的 LangChain 工具實例，並將其傳遞給 LangchainTool 的建構子。

# Instantiate the LangChain tool
tavily_tool_instance = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
    include_answer=True,
    include_raw_content=True,
    include_images=True,
)

# Wrap it with LangchainTool for ADK
adk_tavily_tool = LangchainTool(tool=tavily_tool_instance)

新增至 agent： 在定義時，將包裝過的 LangchainTool 實例加入你的 agent 的 tools 清單中。

from google.adk import Agent

# Define the ADK agent, including the wrapped tool
my_agent = Agent(
    name="langchain_tool_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using TavilySearch.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[adk_tavily_tool] # Add the wrapped tool here
)

完整範例：Tavily 搜尋¶

以下是結合上述步驟，使用 LangChain Tavily 搜尋工具來建立並執行 agent 的完整程式碼。

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.langchain_tool import LangchainTool
from google.genai import types
from langchain_community.tools import TavilySearchResults

# Ensure TAVILY_API_KEY is set in your environment
if not os.getenv("TAVILY_API_KEY"):
    print("Warning: TAVILY_API_KEY environment variable not set.")

APP_NAME = "news_app"
USER_ID = "1234"
SESSION_ID = "session1234"

# Instantiate LangChain tool
tavily_search = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
    include_answer=True,
    include_raw_content=True,
    include_images=True,
)

# Wrap with LangchainTool
adk_tavily_tool = LangchainTool(tool=tavily_search)

# Define Agent with the wrapped tool
my_agent = Agent(
    name="langchain_tool_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using TavilySearch.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[adk_tavily_tool] # Add the wrapped tool here
)

async def setup_session_and_runner():
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=my_agent, app_name=APP_NAME, session_service=session_service)
    return session, runner

# Agent Interaction
async def call_agent_async(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    session, runner = await setup_session_and_runner()
    events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    async for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
await call_agent_async("stock price of GOOG")

2. 使用 CrewAI 工具¶

Agent Development Kit (ADK) 提供了 CrewaiTool 包裝器，可用於整合 CrewAI 函式庫中的 tools。

範例：使用 CrewAI 的 Serper API 進行網頁搜尋¶

Serper API 提供以程式化方式存取 Google Search 結果的能力。這讓應用程式（如 AI agent）能即時執行 Google Search（包含新聞、圖片等），並取得結構化資料，而無需直接爬取網頁內容。

請依照 ADK 安裝與設定指南操作。
安裝相依套件： 安裝所需的 CrewAI tools 套件。例如，若要使用 SerperDevTool：
```
pip install crewai-tools
```
取得 Serper API KEY，並將其設為環境變數。
```
export SERPER_API_KEY=<REPLACE_WITH_API_KEY>
```
匯入： 從 Agent Development Kit (ADK) 匯入 CrewaiTool，以及所需的 CrewAI 工具（例如 SerperDevTool）。
```
from google.adk.tools.crewai_tool import CrewaiTool
from crewai_tools import SerperDevTool
```

實例化與包裝： 建立 CrewAI 工具的實例，並將其傳遞給 CrewaiTool 建構子。重點是，您必須為 ADK 包裝器提供名稱與描述，因為這些資訊會被 Agent Development Kit (ADK) 的底層模型用來判斷何時應該使用該工具。

# Instantiate the CrewAI tool
serper_tool_instance = SerperDevTool(
    n_results=10,
    save_file=False,
    search_type="news",
)

# Wrap it with CrewaiTool for ADK, providing name and description
adk_serper_tool = CrewaiTool(
    name="InternetNewsSearch",
    description="Searches the internet specifically for recent news articles using Serper.",
    tool=serper_tool_instance
)

加入 Agent： 將包裝過的 CrewaiTool 實例加入你的 Agent 的 tools 清單中。

from google.adk import Agent

# Define the ADK agent
my_agent = Agent(
    name="crewai_search_agent",
    model="gemini-2.0-flash",
    description="Agent to find recent news using the Serper search tool.",
    instruction="I can find the latest news for you. What topic are you interested in?",
    tools=[adk_serper_tool] # Add the wrapped tool here
)

完整範例：Serper API¶

以下是結合上述步驟，使用 CrewAI Serper API 搜尋工具來建立並執行 agent 的完整程式碼。

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.crewai_tool import CrewaiTool
from google.genai import types
from crewai_tools import SerperDevTool


# Constants
APP_NAME = "news_app"
USER_ID = "user1234"
SESSION_ID = "1234"

# Ensure SERPER_API_KEY is set in your environment
if not os.getenv("SERPER_API_KEY"):
    print("Warning: SERPER_API_KEY environment variable not set.")

serper_tool_instance = SerperDevTool(
    n_results=10,
    save_file=False,
    search_type="news",
)

adk_serper_tool = CrewaiTool(
    name="InternetNewsSearch",
    description="Searches the internet specifically for recent news articles using Serper.",
    tool=serper_tool_instance
)

serper_agent = Agent(
    name="basic_search_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    # Add the Serper tool
    tools=[adk_serper_tool]
)

# Session and Runner
async def setup_session_and_runner():
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=serper_agent, app_name=APP_NAME, session_service=session_service)
    return session, runner


# Agent Interaction
async def call_agent_async(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    session, runner = await setup_session_and_runner()
    events = runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    async for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

# Note: In Colab, you can directly use 'await' at the top level.
# If running this code as a standalone Python script, you'll need to use asyncio.run() or manage the event loop.
await call_agent_async("what's the latest news on AI Agents?")