Open5

Google ADK (Agent Development Kit) MCP の例

knaka Tech-Blogknaka Tech-Blog

概要

  • Google , Agent Development Kit + MCP 試す例になります。
  • 外部API連携で結果を、画面に表示
  • python 版

[ 公開 2025/07/26 ]


  • 前の ADK 続編になります。

https://zenn.dev/link/comments/c2ebf2aa47cb0b


環境

  • Agent Development Kit
  • Python 3.13.4
  • LLM= gemini

書いたコード

https://gist.github.com/kuc-arc-f/f1e5d7bd8636a0512b8d2ad60a1d9a6c


  • dev-start
adk web

  • tool : 外部API連携
  • json 読み込む例です。画面に表形式の表示
  • get_todo_list
def get_todo_list(limit: int = 5) -> dict:
    """
    外部APIを呼び出してToDoリストを取得します。

    Args:
        limit (int): 取得するToDoの最大数。デフォルトは5。

    Returns:
        dict: ステータスとToDoリストの結果。
              例: {"status": "success", "todos": [{"id": 1, "title": "...", "completed": False}, ...]}
    """
    try:
        # JSONPlaceholderのToDo APIを呼び出し
        response = requests.get(f"https://jsonplaceholder.typicode.com/todos?_limit={limit}", timeout=10)
        response.raise_for_status() # ステータスコードが200でない場合に例外を発生

        todos = response.json()

        # 必要な情報のみ抽出してリスト化
        todo_list = [
            {
                "id": todo["id"],
                "title": todo["title"],
                "completed": todo["completed"]
            }
            for todo in todos
        ]

        return {
            "status": "success",
            "todos": todo_list
        }
    except requests.exceptions.RequestException as e:
        # エラー発生時の処理
        return {
            "status": "error",
            "message": f"API呼び出し中にエラーが発生しました: {str(e)}"
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"予期しないエラーが発生しました: {str(e)}"
        }


  • プロンプト-1
今日の日付 わかります?
  • プロンプト-2
外部APIからのTODOリスト 取得して欲しい。TODO項目は Markdown(表形式)です。最大 5件 表示したい。

knaka Tech-Blogknaka Tech-Blog

API連携 登録処理 , Google ADK + MCP

  • 外部API連携で、DB 登録
  • 入力から、自然言語処理で。必要なパラメータ抽出し
  • 登録API POST通信で。登録

環境

  • Agent Development Kit
  • Python 3.13.4
  • LLM= gemini

書いたコード

https://gist.github.com/kuc-arc-f/bb4f24f3efea0a41100beed0dce21b2f


  • dev-start
adk web

  • tool : 外部API連携
  • register_item
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実行のシミュレーションです ---


  • プロンプト-1
お茶 120 JPY をAPI送信して欲しい。


  • d1 databaseに。登録できました。

knaka Tech-Blogknaka Tech-Blog

React 操作画面 node , Google ADK + MCP

  • React client 操作画面の追加。
  • node , ADK間は。POST通信です。

環境

  • Agent Development Kit
  • Python 3.13.4
  • LLM= gemini
  • node22
  • React

書いたコード

https://github.com/kuc-arc-f/rolldown_2ex/tree/main/mcp_react6


  • dev-start
npm run dev

  • .env
  • ADKの接続先
ADK_API_URL="http://localhost:8000"

  • 操作画面
  • agent: tool_agent_3 対象
  • mcp_react6/src/client/Home.tsx

https://github.com/kuc-arc-f/rolldown_2ex/blob/main/mcp_react6/src/client/Home.tsx

const chatStart = async function(){
    try{    
      setText("");
      setIsLoading(false);
      const elem = document.getElementById("input_text") as HTMLInputElement;
      let inText = "";
      let inToolText = "";
      if(elem){
        inText = elem.value;
      };
      const elemToolText = document.getElementById("tool_text") as HTMLInputElement;
      if(elemToolText){
        inToolText = elemToolText.value;
      }

      console.log("inText=", inText);
      console.log("inToolText=", inToolText);
      if(!inText){ return; }
      setIsLoading(true);
      const agentName = AgentUtil.validAgentName(inText.trim());
      console.log("agentName=", agentName);
      if(agentName){
        let items = [];
        if(agentName === "tool_agent_3"){
          items = toolAgent3();
        }
        console.log(items);

        let htmAll = "";
        let itemCount = 1;
        const agentSendProc = async function(){
          for(const row of items) {
            htmAll += `<div class="label character-label w-full">${row.title}</div>`;
            console.log(row);
            let res = null;
            if(itemCount === 2){
              res = await AgentUtil.postAgent(inToolText, "tool_agent_3");
            }else{
              res = await AgentUtil.postAgent(row.text, "tool_agent_3");
            }
            itemCount += 1; 
            console.log(res);
            htmAll += `<div class="chat-bubble character-bubble w-full">`;
            htmAll += marked.parse(res.text);
            htmAll +=  `</div>`
            //console.log(htmAll);
            setText(htmAll);
          };
          setIsLoading(false);
        }
        agentSendProc();
        return;
      }
    } catch(e){
      console.error(e);
    }
  }


knaka Tech-Blogknaka Tech-Blog

Slack メッセージ送信 , Google ADK + MCP

  • Slack メッセージ送信、Webhook使用

環境

  • Agent Development Kit
  • Python 3.13.4
  • LLM= gemini
  • node22
  • React

書いたコード

https://gist.github.com/kuc-arc-f/bbffd879953be926782a0b630eb03592

  • 入力から、自然言語処理で。メッセージを抽出する
  • メッセージを、Webhook経由で送信する。

  • 準備: geminiに質問すると。テキスト送信できました。
  • Slack Appの作成, Incoming Webhooksを有効 , Webhook URL が表示される。
  • 参考のWebhook URL を、実行すると。メッセージ送信できました。

  • 入力text
こんにちは のメッセージを Slackに送信して欲しい。

  • agent.py
  • tools 内で、Slack送信機能を追加

# --------------------------------------------------------------------------
# Toolの定義
# --------------------------------------------------------------------------
def slack_send_message(message: str) -> str:
    """
    Slackにメッセージを送信する実処理。
    LLMはこのdocstringを読んで、'message'という引数が必要だと理解します。

    指定された メッセージ Slackに送信します。
    自然言語から「メッセージ」を自動で抽出して、この関数の引数に渡します。

    Args:
        message: Slackに送信するテキストメッセージ。

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

    # --- ここからが外部API実行のシミュレーションです ---
    # TODO: あなたの実際のAPIエンドポイントURLに置き換えてください
 
    payload = {"text": message}

    print(f"  [Webhook送信] URL: {Webhook_URL}")
    print(f"  [Webhook送信] データ: {payload}")

    try:
        # 実際のAPI呼び出しを行う場合は、以下のコメントを解除します。
        response = requests.post(Webhook_URL, json=payload)
        print("status_code=" + str(response.status_code))

        # このサンプルでは、API呼び出しが成功したと仮定します。
        print("  [Webhook結果] 成功したと仮定します。")
        # Slackにメッセージを送信する実処理
        return f"承知いたしました。「{message}」をSlackにメッセージを送信しました。"

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

knaka Tech-Blogknaka Tech-Blog

discord メッセージ送信 , Google ADK + MCP

  • discord メッセージ送信、Webhook使用

環境

  • Agent Development Kit
  • Python 3.13.4
  • LLM= gemini
  • discord web版

書いたコード

https://gist.github.com/kuc-arc-f/ab37495a7128ce7c010843512bd3d8f7

  • 入力から、自然言語処理で。メッセージを抽出する
  • メッセージを、Webhook経由で送信する。

  • Webhook-URL 取得方法
  • サーバー設定, 連携サービス, ウェブフック


  • text送信 例

  • 入力text
こんにちは  のメッセージを discordに送信して欲しい。

  • agent.py
  • tools 内で、discord 送信機能を追加
def discord_send_message(message: str) -> str:
    """
    discordにメッセージを送信する実処理。
    LLMはこのdocstringを読んで、'message'という引数が必要だと理解します。

    指定された メッセージ discordに送信します。
    自然言語から「メッセージ」を自動で抽出して、この関数の引数に渡します。

    Args:
        message: discordに送信するテキストメッセージ。

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

    # --- ここからが外部API実行のシミュレーションです ---
    # TODO: あなたの実際のAPIエンドポイントURLに置き換えてください
 
    # 送信するメッセージ
    payload = {
        "content": message,
        "username": "Webhook-user-2", # 送信者の名前(任意)
    }

    print(f"  [Webhook送信] URL: {WEBHOOK_URL}")
    print(f"  [Webhook送信] データ: {payload}")

    try:
        # 実際のAPI呼び出しを行う場合は、以下のコメントを解除します。
        # HTTP POSTリクエストを送信
        response = requests.post(
            WEBHOOK_URL,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"}
        )
        print("status_code=" + str(response.status_code))

        # このサンプルでは、API呼び出しが成功したと仮定します。
        print("  [Webhook結果] 成功したと仮定します。")
        # メッセージを送信する実処理
        return f"承知いたしました。「{message}」をメッセージを送信しました。"

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