Google ADK , MCP example, tool_agent_3

.env

GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY="key"

__init__.py

from . import agent

agent.py

import datetime
import requests
from zoneinfo import ZoneInfo
from google.adk.agents import Agent

WORKERS_API_URL="https://localhost"

# --------------------------------------------------------------------------
# Toolの定義
# --------------------------------------------------------------------------
def register_item(item_name: str, value: int) -> str:
    """
    指定された項目名と数値を外部APIに送信して登録します。
    自然言語から「項目名」と「数値」を自動で抽出して、この関数の引数に渡します。

    Args:
        item_name (str): 登録する項目名。 (例: "売上", "在庫数")
        value (int): 登録する数値。 (例: 10000, 50)

    Returns:
        str: API実行結果を要約した、ユーザーへの返答メッセージ。
    """
    print("--- Tool: register_item が呼び出されました ---")
    print(f"  [抽出された引数] item_name='{item_name}', value={value}")

    # このサンプルでは、API呼び出しが成功したと仮定します。
    #print("  [API結果] 成功したと仮定します。")
    # ユーザーへの返答メッセージを生成します。
    #return f"承知いたしました。「{item_name}」のデータとして「{value}」をシステムに登録しました。"

    # --- ここからが外部API実行のシミュレーションです ---
    # TODO: あなたの実際のAPIエンドポイントURLに置き換えてください
    api_url = WORKERS_API_URL + "/api/mcp_use_price/create"
    # const item = {title: text, price: num }

    payload = {
        "title": item_name,
        "price": value,
    }
    #    "source": "adk-agent"
    print(f"  [API送信] URL: {api_url}")
    print(f"  [API送信] データ: {payload}")

    try:
        # 実際のAPI呼び出しを行う場合は、以下のコメントを解除します。
        # headers = {"Authorization": "Bearer YOUR_API_KEY"} # 必要に応じてヘッダーを追加
        # response = requests.post(api_url, json=payload, headers=headers)
        # response.raise_for_status()  # HTTPステータスコードが2xxでない場合に例外を発生
        response = requests.post(api_url, json=payload)
        print("status_code=" + str(response.status_code))

        # このサンプルでは、API呼び出しが成功したと仮定します。
        print("  [API結果] 成功したと仮定します。")
        # ユーザーへの返答メッセージを生成します。
        return f"承知いたしました。「{item_name}」のデータとして「{value}」をシステムに登録しました。"

    except requests.exceptions.RequestException as e:
        # API呼び出しでネットワークエラーなどが発生した場合の処理
        print(f"  [API結果] エラーが発生しました: {e}")
        return f"申し訳ありません。システムの通信エラーにより、「{item_name}」の登録に失敗しました。"
    except Exception as e:
        # その他の予期せぬエラーが発生した場合の処理
        print(f"  [API結果] 予期せぬエラーが発生しました: {e}")
        return f"申し訳ありません。予期せぬエラーにより、「{item_name}」の登録に失敗しました。"
    finally:
        print("--- Tool: 処理を終了します ---")
    # --- ここまでが外部API実行のシミュレーションです ---

# get_now_date
# Returns the current date in YYYY-MM-DD format
def get_now_date() -> str:
    """Returns the current date in YYYY-MM-DD format.

    Returns:
        dict: status and result with current date.
    """
    current_date = datetime.datetime.now().strftime("%Y-%m-%d")
    return f"今日の日付: {current_date}"

# Agent
root_agent = Agent(
    name="tool_agent_3",
    model="gemini-2.0-flash",
    description=(
        "Agent の回答は。 本日の日時、 ユーザーの自然言語から項目と数値を抽出し、外部APIに登録します。"
    ),
    instruction=(
        "You are a helpful agent.  本日の日時 を返却します。"
        "ユーザーの入力から「項目名」と「数値」を抽出し、外部APIに送信してください。"
        "例: 「温度は36.5度です」→ item: 温度, value: 36.5"
    ),
    tools=[get_now_date, register_item],
)