การเริ่มต้นใช้งาน MCP, ADK และ A2A

1. ภาพรวม

เอเจนต์ AI กำลังได้รับความนิยมอย่างรวดเร็ว ซึ่งปฏิวัติการทำงานอัตโนมัติและการตัดสินใจด้วยความสามารถในการทำงานด้วยตนเอง เรียนรู้ และโต้ตอบกับสภาพแวดล้อมเพื่อบรรลุเป้าหมาย

แต่เราจะสร้างเอเจนต์ได้อย่างไร Codelab นี้จะช่วยให้คุณเริ่มต้นใช้งานได้โดยแสดงวิธีสร้างเอเจนต์สกุลเงินที่สามารถแปลงสกุลเงินของประเทศต่างๆ โดยมีเป้าหมายเพื่ออธิบายเทคโนโลยีล่าสุดเพื่อช่วยให้คุณเข้าใจตัวย่อต่างๆ ที่อาจเห็นในอินเทอร์เน็ต (MCP, ADK, A2A)

สถาปัตยกรรม

Model Context Protocol (MCP)

โปรโตคอลบริบทของโมเดล (MCP) เป็นโปรโตคอลแบบเปิดที่กำหนดมาตรฐานวิธีที่แอปพลิเคชันให้บริบทแก่ LLM MCP มีวิธีมาตรฐานในการเชื่อมต่อโมเดล AI กับทรัพยากร พรอมต์ และเครื่องมือ

ชุดพัฒนาเอเจนต์ (ADK)

Agent Development Kit (ADK) คือเฟรมเวิร์กการจัดระเบียบที่ยืดหยุ่นสำหรับการพัฒนาและติดตั้งใช้งานเอเจนต์ AI ADK ไม่ขึ้นอยู่กับโมเดล ไม่ขึ้นอยู่กับการติดตั้งใช้งาน และสร้างขึ้นเพื่อให้ใช้งานร่วมกับเฟรมเวิร์กอื่นๆ ได้ ADK ออกแบบมาเพื่อให้การพัฒนาเอเจนต์มีความคล้ายกับการพัฒนาซอฟต์แวร์มากขึ้น เพื่อช่วยให้นักพัฒนาแอปสร้าง ปรับใช้ และจัดระเบียบสถาปัตยกรรมเอเจนต์ได้ง่ายขึ้น ซึ่งครอบคลุมตั้งแต่การทำงานง่ายๆ ไปจนถึงเวิร์กโฟลว์ที่ซับซ้อน

โปรโตคอล Agent2Agent (A2A)

โปรโตคอล Agent2Agent (A2A) เป็นมาตรฐานแบบเปิดที่ออกแบบมาเพื่อช่วยให้ตัวแทน AI สื่อสารและทำงานร่วมกันได้อย่างราบรื่น เช่นเดียวกับที่ MCP มอบวิธีมาตรฐานในการให้สิทธิ์ LLM เข้าถึงข้อมูลและเครื่องมือ A2A ก็มอบวิธีมาตรฐานให้เอเจนต์พูดคุยกับเอเจนต์อื่นๆ ในโลกที่เอเจนต์สร้างขึ้นโดยใช้เฟรมเวิร์กที่หลากหลายและโดยผู้ให้บริการที่แตกต่างกัน A2A จะเป็นภาษาที่ใช้ร่วมกัน ซึ่งจะช่วยทำลายไซโลและส่งเสริมการทำงานร่วมกัน

สิ่งที่คุณจะได้เรียนรู้

  • วิธีสร้างเซิร์ฟเวอร์ MCP ในเครื่อง
  • การทำให้เซิร์ฟเวอร์ MCP ใช้งานได้กับ Cloud Run
  • วิธีสร้างเอเจนต์ด้วย Agent Development Kit ที่ใช้เครื่องมือ MCP
  • วิธีเปิดเผยตัวแทน ADK เป็นเซิร์ฟเวอร์ A2A
  • การทดสอบเซิร์ฟเวอร์ A2A โดยใช้ไคลเอ็นต์ A2A

สิ่งที่คุณต้องมี

  • เบราว์เซอร์ เช่น Chrome หรือ Firefox
  • โปรเจ็กต์ Google Cloud ที่เปิดใช้การเรียกเก็บเงิน

2. ก่อนเริ่มต้น

สร้างโปรเจ็กต์

  1. ใน Google Cloud Console ให้เลือกหรือสร้างโปรเจ็กต์ Google Cloud ในหน้าตัวเลือกโปรเจ็กต์
  2. ตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินสำหรับโปรเจ็กต์ Cloud แล้ว ดูวิธีตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินในโปรเจ็กต์แล้วหรือไม่
  3. เปิดใช้งาน Cloud Shell โดยคลิกลิงก์นี้ คุณสลับระหว่างเทอร์มินัล Cloud Shell (สําหรับเรียกใช้คําสั่งคลาวด์) กับ Editor (สําหรับสร้างโปรเจ็กต์) ได้โดยคลิกปุ่มที่เกี่ยวข้องจาก Cloud Shell
  4. เมื่อเชื่อมต่อกับ Cloud Shell แล้ว ให้ตรวจสอบว่าคุณได้รับการตรวจสอบสิทธิ์แล้วและตั้งค่าโปรเจ็กต์เป็นรหัสโปรเจ็กต์ของคุณโดยใช้คำสั่งต่อไปนี้
gcloud auth list
  1. เรียกใช้คำสั่งต่อไปนี้ใน Cloud Shell เพื่อยืนยันว่าคำสั่ง gcloud รู้จักโปรเจ็กต์ของคุณ
gcloud config list project
  1. ใช้คำสั่งต่อไปนี้เพื่อตั้งค่าโปรเจ็กต์
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
  1. เปิดใช้ API ที่จำเป็นโดยใช้คำสั่งต่อไปนี้ การดำเนินการนี้อาจใช้เวลาสักครู่
gcloud services enable cloudresourcemanager.googleapis.com \
                       servicenetworking.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       artifactregistry.googleapis.com \
                       aiplatform.googleapis.com \
                       compute.googleapis.com
  1. ตรวจสอบว่าคุณมี Python 3.10 ขึ้นไป

โปรดดูคำสั่งและการใช้งาน gcloud ในเอกสารประกอบ

3. การติดตั้ง

  1. โคลนที่เก็บ
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
  1. ติดตั้ง uv (ใช้เพื่อจัดการทรัพยากร Dependency)
# macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (uncomment below line)
# powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
  1. กำหนดค่าตัวแปรสภาพแวดล้อม (ผ่านไฟล์ .env):

สร้างไฟล์ .env โดยเรียกใช้คำสั่งต่อไปนี้

echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env

4. สร้างเซิร์ฟเวอร์ MCP ในเครื่อง

ก่อนที่จะจัดระเบียบเอเจนต์สกุลเงิน คุณจะต้องสร้างเซิร์ฟเวอร์ MCP เพื่อเปิดเผยเครื่องมือที่เอเจนต์ของคุณจะต้องใช้

เซิร์ฟเวอร์ MCP ช่วยให้คุณเขียนโปรแกรมที่มีขนาดเล็กเพื่อแสดงความสามารถเฉพาะ (เช่น การดึงอัตราแลกเปลี่ยนสกุลเงิน) เป็นเครื่องมือได้ จากนั้นเอเจนต์หรือเอเจนต์หลายรายจะเข้าถึงเครื่องมือเหล่านี้ได้โดยใช้โปรโตคอลบริบทของโมเดล (MCP) ที่ได้มาตรฐาน

คุณสามารถใช้แพ็กเกจ Python FastMCP เพื่อสร้างเซิร์ฟเวอร์ MCP ที่แสดงเครื่องมือเดียวที่ชื่อ get_exchange_rate get_exchange_rate เครื่องมือจะทำการเรียกผ่านอินเทอร์เน็ตไปยัง Frankfurter API เพื่อรับอัตราแลกเปลี่ยนปัจจุบันระหว่าง 2 สกุลเงิน

คุณดูโค้ดสำหรับเซิร์ฟเวอร์ MCP ได้ในไฟล์ mcp-server/server.py

import logging
import os

import httpx
from fastmcp import FastMCP

# Set up logging
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

mcp = FastMCP("Currency MCP Server 💵")

@mcp.tool()
def get_exchange_rate(
    currency_from: str = 'USD',
    currency_to: str = 'EUR',
    currency_date: str = 'latest',
):
    """Use this to get current exchange rate.

    Args:
        currency_from: The currency to convert from (e.g., "USD").
        currency_to: The currency to convert to (e.g., "EUR").
        currency_date: The date for the exchange rate or "latest". Defaults to "latest".

    Returns:
        A dictionary containing the exchange rate data, or an error message if the request fails.
    """
    logger.info(f"--- 🛠️ Tool: get_exchange_rate called for converting {currency_from} to {currency_to} ---")
    try:
        response = httpx.get(
            f'https://api.frankfurter.app/{currency_date}',
            params={'from': currency_from, 'to': currency_to},
        )
        response.raise_for_status()

        data = response.json()
        if 'rates' not in data:
            return {'error': 'Invalid API response format.'}
        logger.info(f'✅ API response: {data}')
        return data
    except httpx.HTTPError as e:
        return {'error': f'API request failed: {e}'}
    except ValueError:
        return {'error': 'Invalid JSON response from API.'}

if __name__ == "__main__":
    logger.info(f"🚀 MCP server started on port {os.getenv('PORT', 8080)}")
    # Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
    asyncio.run(
        mcp.run_async(
            transport="streamable-http",
            host="0.0.0.0",
            port=os.getenv("PORT", 8080),
        )
    )

หากต้องการเริ่มเซิร์ฟเวอร์ MCP ในเครื่อง ให้เปิดเทอร์มินัลแล้วเรียกใช้คำสั่งต่อไปนี้ (เซิร์ฟเวอร์จะเริ่มทำงานใน http://localhost:8080)

uv run mcp-server/server.py

ทดสอบว่าเซิร์ฟเวอร์ MCP ทำงานอย่างถูกต้องและเข้าถึงเครื่องมือ get_exchange_rate ได้โดยใช้ Model Context Protocol

ในหน้าต่างเทอร์มินัลใหม่ (เพื่อไม่ให้หยุดเซิร์ฟเวอร์ MCP ในเครื่อง) ให้เรียกใช้คำสั่งต่อไปนี้

uv run mcp-server/test_server.py

คุณควรเห็นอัตราแลกเปลี่ยนปัจจุบันของ 1 USD (ดอลลาร์สหรัฐ) เป็น EUR (ยูโร) ดังนี้

--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
---  Success: {
  "amount": 1.0,
  "base": "USD",
  "date": "2025-05-26",
  "rates": {
    "EUR": 0.87866
  }
} ---

ยอดเยี่ยม! คุณมีเซิร์ฟเวอร์ MCP ที่ใช้งานได้พร้อมเครื่องมือที่ตัวแทนจะเข้าถึงได้

ก่อนที่จะไปยังสถานีถัดไป ให้หยุดเซิร์ฟเวอร์ MCP ที่ทำงานในเครื่องโดยเรียกใช้ Ctrl+C (หรือ Command+C ใน Mac) ในเทอร์มินัลที่คุณเริ่มต้น

5. ทำให้เซิร์ฟเวอร์ MCP ใช้งานได้ใน Cloud Run

ตอนนี้คุณพร้อมที่จะติดตั้งใช้งานเซิร์ฟเวอร์ MCP เป็นเซิร์ฟเวอร์ MCP ระยะไกลใน Cloud Run แล้ว 🚀☁️

ประโยชน์ของการเรียกใช้เซิร์ฟเวอร์ MCP จากระยะไกล

การเรียกใช้เซิร์ฟเวอร์ MCP จากระยะไกลใน Cloud Run มีประโยชน์หลายประการ ดังนี้

  • 📈ความสามารถในการปรับขนาด: Cloud Run สร้างขึ้นเพื่อปรับขนาดออกอย่างรวดเร็วเพื่อจัดการคำขอขาเข้าทั้งหมด Cloud Run จะปรับขนาดเซิร์ฟเวอร์ MCP โดยอัตโนมัติตามความต้องการ
  • 👥เซิร์ฟเวอร์ส่วนกลาง: คุณแชร์สิทธิ์เข้าถึงเซิร์ฟเวอร์ MCP ส่วนกลางกับสมาชิกในทีมได้ผ่านสิทธิ์ IAM ซึ่งจะช่วยให้สมาชิกเชื่อมต่อกับเซิร์ฟเวอร์จากเครื่องในพื้นที่ของตนเองได้แทนที่จะต้องเรียกใช้เซิร์ฟเวอร์ของตนเองในเครื่อง หากมีการเปลี่ยนแปลงในเซิร์ฟเวอร์ MCP สมาชิกในทีมทุกคนจะได้รับประโยชน์จากการเปลี่ยนแปลงนั้น
  • 🔐ความปลอดภัย: Cloud Run มีวิธีง่ายๆ ในการบังคับใช้คำขอที่ผ่านการตรวจสอบสิทธิ์ ซึ่งจะอนุญาตเฉพาะการเชื่อมต่อที่ปลอดภัยกับเซิร์ฟเวอร์ MCP เท่านั้น เพื่อป้องกันการเข้าถึงที่ไม่ได้รับอนุญาต

เปลี่ยนเป็นไดเรกทอรี mcp-server โดยใช้คำสั่งต่อไปนี้

cd mcp-server

ทำให้เซิร์ฟเวอร์ MCP ใช้งานได้กับ Cloud Run

gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .

หากบริการของคุณได้รับการติดตั้งใช้งานเรียบร้อยแล้ว คุณจะเห็นข้อความต่อไปนี้

Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.

การตรวจสอบสิทธิ์ไคลเอ็นต์ MCP

เนื่องจากคุณระบุ --no-allow-unauthenticated เพื่อกำหนดให้มีการตรวจสอบสิทธิ์ ไคลเอ็นต์ MCP ใดก็ตามที่เชื่อมต่อกับเซิร์ฟเวอร์ MCP ระยะไกลจะต้องตรวจสอบสิทธิ์

เอกสารอย่างเป็นทางการสำหรับโฮสต์เซิร์ฟเวอร์ MCP ใน Cloud Run มีข้อมูลเพิ่มเติมเกี่ยวกับหัวข้อนี้โดยขึ้นอยู่กับตำแหน่งที่คุณเรียกใช้ไคลเอ็นต์ MCP

คุณจะต้องเรียกใช้พร็อกซี Cloud Run เพื่อสร้างอุโมงค์ที่ตรวจสอบสิทธิ์แล้วไปยังเซิร์ฟเวอร์ MCP ระยะไกลในเครื่อง

โดยค่าเริ่มต้น URL ของบริการ Cloud Run จะกำหนดให้คำขอทั้งหมดต้องได้รับอนุญาตด้วยบทบาท IAM ของผู้เรียกใช้ Cloud Run (roles/run.invoker) การเชื่อมโยงนโยบาย IAM นี้ช่วยให้มั่นใจได้ว่าจะใช้กลไกการรักษาความปลอดภัยที่รัดกุมเพื่อตรวจสอบสิทธิ์ไคลเอ็นต์ MCP ในเครื่อง

คุณควรตรวจสอบว่าคุณหรือสมาชิกในทีมที่พยายามเข้าถึงเซิร์ฟเวอร์ MCP ระยะไกลมีroles/run.invokerบทบาท IAM ที่เชื่อมโยงกับหลักการ IAM (บัญชี Google Cloud) ของตน

gcloud run services proxy mcp-server --region=us-central1

คุณควรเห็นเอาต์พุตต่อไปนี้

Proxying to Cloud Run service [mcp-server] in project [<YOUR_PROJECT_ID>] region [us-central1]
http://127.0.0.1:8080 proxies to https://mcp-server-abcdefgh-uc.a.run.app

ตอนนี้การรับส่งข้อมูลทั้งหมดไปยัง http://127.0.0.1:8080 จะได้รับการตรวจสอบสิทธิ์และส่งต่อไปยังเซิร์ฟเวอร์ MCP ระยะไกล

ทดสอบเซิร์ฟเวอร์ MCP ระยะไกล

ในเทอร์มินัลใหม่ ให้กลับไปที่โฟลเดอร์รูทและเรียกใช้ไฟล์ mcp-server/test_server.py อีกครั้งเพื่อให้แน่ใจว่าเซิร์ฟเวอร์ MCP ระยะไกลทำงานอยู่

cd ..
uv run mcp-server/test_server.py

คุณควรเห็นเอาต์พุตที่คล้ายกับตอนที่เรียกใช้เซิร์ฟเวอร์ในเครื่อง

--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
---  Success: {
  "amount": 1.0,
  "base": "USD",
  "date": "2025-05-26",
  "rates": {
    "EUR": 0.87866
  }
} ---

คุณสามารถค้นหาบันทึกของเซิร์ฟเวอร์ MCP ของ Cloud Run ที่ติดตั้งใช้งานแล้วได้หากต้องการยืนยันว่ามีการเรียกเซิร์ฟเวอร์ระยะไกลจริง

gcloud run services logs read mcp-server --region us-central1 --limit 5

คุณควรเห็นเอาต์พุตต่อไปนี้ในบันทึก

2025-06-04 14:28:29,871 [INFO]: --- 🛠️ Tool: get_exchange_rate called for converting USD to EUR ---
2025-06-04 14:28:30,610 [INFO]: HTTP Request: GET https://api.frankfurter.app/latest?from=USD&to=EUR "HTTP/1.1 200 OK"
2025-06-04 14:28:30,611 [INFO]:  API response: {'amount': 1.0, 'base': 'USD', 'date': '2025-06-03', 'rates': {'EUR': 0.87827}}

เมื่อมีเซิร์ฟเวอร์ MCP ระยะไกลแล้ว คุณก็สามารถสร้างเอเจนต์ได้เลย 🤖

6. สร้าง Agent ด้วย Agent Development Kit (ADK)

คุณมีเซิร์ฟเวอร์ MCP ที่ใช้งานแล้ว ตอนนี้ก็ถึงเวลาสร้างเอเจนต์สกุลเงินโดยใช้ชุดพัฒนาเอเจนต์ (ADK)

Agent Development Kit เพิ่งเปิดตัวเวอร์ชันเสถียร v1.0.0 เหตุการณ์สำคัญนี้แสดงให้เห็นว่าตอนนี้ Python ADK พร้อมใช้งานจริงแล้ว โดยมีแพลตฟอร์มที่เชื่อถือได้และมีประสิทธิภาพสำหรับนักพัฒนาแอปในการสร้างและติดตั้งใช้งานเอเจนต์ในสภาพแวดล้อมจริงได้อย่างมั่นใจ

ADK ช่วยให้สร้างเอเจนต์ที่มีขนาดเล็กมากและเชื่อมต่อกับเซิร์ฟเวอร์ MCP ได้อย่างง่ายดายด้วยการรองรับเครื่องมือ MCP ในตัว เอเจนต์สกุลเงินจะเข้าถึงเครื่องมือ get_exchange_rate โดยใช้คลาส MCPToolset ของ ADK

รหัสสำหรับตัวแทนสกุลเงินอยู่ใน currency_agent/agent.py

import logging
import os

from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams

logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

load_dotenv()

SYSTEM_INSTRUCTION = (
    "You are a specialized assistant for currency conversions. "
    "Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
    "If the user asks about anything other than currency conversion or exchange rates, "
    "politely state that you cannot help with that topic and can only assist with currency-related queries. "
    "Do not attempt to answer unrelated questions or use tools for other purposes."
)

def create_agent() -> LlmAgent:
    """Constructs the ADK currency conversion agent."""
    logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
    logger.info("--- 🤖 Creating ADK Currency Agent... ---")
    return LlmAgent(
        model="gemini-2.5-flash",
        name="currency_agent",
        description="An agent that can help with currency conversions",
        instruction=SYSTEM_INSTRUCTION,
        tools=[
            MCPToolset(
                connection_params=StreamableHTTPConnectionParams(
                    url=os.getenv("MCP_SERVER_URL", "http://localhost:8080/mcp")
                )
            )
        ],
    )


root_agent = create_agent()

หากต้องการทดสอบเอเจนต์สกุลเงินอย่างรวดเร็ว คุณสามารถใช้ประโยชน์จาก UI สำหรับนักพัฒนาซอฟต์แวร์ของ ADK ได้โดยเรียกใช้ adk web:

uv run adk web

ในเบราว์เซอร์ ให้ไปที่ http://localhost:8000 เพื่อดูและทดสอบเอเจนต์

ตรวจสอบว่าได้เลือก currency_agent เป็นตัวแทนที่มุมซ้ายบนของเว็บ UI

UI บนเว็บของ ADK

ถามตัวแทนในแชท เช่น "250 CAD เท่ากับกี่ USD" คุณควรเห็นตัวแทนโทรหาเครื่องมือ get_exchange_rate MCP ของเราก่อนที่ตัวแทนจะตอบ

Agent สกุลเงินของ ADK Web

Agent ทำงานแล้ว โดยสามารถจัดการคำค้นหาที่เกี่ยวข้องกับการแปลงสกุลเงิน 💸

7. โปรโตคอล Agent2Agent (A2A)

โปรโตคอล Agent2Agent (A2A) เป็นมาตรฐานแบบเปิดที่ออกแบบมาเพื่อช่วยให้ตัวแทน AI สื่อสารและทำงานร่วมกันได้อย่างราบรื่น ซึ่งช่วยให้เอเจนต์ที่สร้างขึ้นโดยใช้เฟรมเวิร์กที่หลากหลายและโดยผู้ให้บริการที่แตกต่างกันสามารถสื่อสารกันในภาษาเดียวกันได้ ซึ่งจะช่วยลดการทำงานแบบแยกส่วนและส่งเสริมการทำงานร่วมกัน

โปรโตคอล A2A

A2A ช่วยให้ตัวแทนทำสิ่งต่อไปนี้ได้

  • ค้นพบ: ค้นหาเอเจนต์อื่นๆ และเรียนรู้ทักษะ (AgentSkill) และความสามารถ (AgentCapabilities) ของเอเจนต์เหล่านั้นโดยใช้การ์ดเอเจนต์ที่ได้มาตรฐาน
  • สื่อสาร: แลกเปลี่ยนข้อความและข้อมูลอย่างปลอดภัย
  • ทำงานร่วมกัน: มอบหมายงานและประสานงานเพื่อบรรลุเป้าหมายที่ซับซ้อน

โปรโตคอล A2A ช่วยให้การสื่อสารนี้เป็นไปได้ผ่านกลไกต่างๆ เช่น "การ์ดเอเจนต์" ซึ่งทำหน้าที่เป็นนามบัตรดิจิทัลที่เอเจนต์ใช้เพื่อโฆษณาความสามารถและข้อมูลการเชื่อมต่อของตนได้

บัตรตัวแทน A2A

ตอนนี้ถึงเวลาเปิดเผยเอเจนต์สกุลเงินโดยใช้ A2A เพื่อให้เอเจนต์และไคลเอ็นต์อื่นๆ เรียกใช้ได้

A2A Python SDK

A2A Python SDK มีโมเดล Pydantic สำหรับทรัพยากรแต่ละรายการที่กล่าวถึงข้างต้น ได้แก่ AgentSkill, AgentCapabilities และ AgentCard ซึ่งมีอินเทอร์เฟซสำหรับการเร่งการพัฒนาและการผสานรวมกับโปรโตคอล A2A

AgentSkill คือวิธีที่คุณจะโฆษณาต่อตัวแทนอื่นๆ ว่าตัวแทนสกุลเงินมีเครื่องมือสำหรับ get_exchange_rate:

# A2A Agent Skill definition
skill = AgentSkill(
    id='get_exchange_rate',
    name='Currency Exchange Rates Tool',
    description='Helps with exchange values between various currencies',
    tags=['currency conversion', 'currency exchange'],
    examples=['What is exchange rate between USD and GBP?'],
)

จากนั้นในส่วนของ AgentCard จะแสดงทักษะและความสามารถของเอเจนต์พร้อมกับรายละเอียดเพิ่มเติม เช่น โหมดอินพุตและเอาต์พุตที่เอเจนต์จัดการได้

# A2A Agent Card definition
agent_card = AgentCard(
    name='Currency Agent',
    description='Helps with exchange rates for currencies',
    url=f'http://{host}:{port}/',
    version='1.0.0',
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)

อินเทอร์เฟซ AgentExecutor จะจัดการตรรกะหลักของวิธีที่เอเจนต์ A2A ประมวลผลคำขอและสร้างการตอบกลับ/เหตุการณ์ A2A Python SDK มีคลาสฐานแบบนามธรรม a2a.server.agent_execution.AgentExecutor ที่คุณต้องติดตั้งใช้งาน

ถึงเวลาแล้วที่จะนำทุกอย่างมารวมกันด้วยเอเจนต์สกุลเงินและแสดงให้เห็นถึงประสิทธิภาพของ A2A

8. เซิร์ฟเวอร์ Currency Agent A2A

ตอนนี้คุณจะมาดูโค้ดบางส่วนและดูว่าโค้ดส่วนต่างๆ ที่ประกอบกันเป็นเซิร์ฟเวอร์ A2A นั้นทำงานร่วมกันอย่างไร

เมื่อดูภายในไฟล์ currency_agent/agent_executor.py คุณจะเห็นคลาส ADKAgentExecutor ซึ่งรับค่ามาจากคลาส AgentExecutor ที่เป็นนามธรรมของ A2A โดยจะจัดการการเรียกใช้ตัวแทน ADK ด้วยการเรียกใช้โปรแกรมเรียกใช้ ADK, การประมวลผลคำขอไปยังตัวแทน และการแปลงไปมาระหว่าง google.genai.types ซึ่ง ADK ใช้ และ a2a.types ซึ่ง A2A ใช้

# ... see file for full code

class ADKAgentExecutor(AgentExecutor):
    """An AgentExecutor that runs an ADK agent."""

    def __init__(self, runner: Runner, card: AgentCard):
        self.runner = runner
        self._card = card
        self._running_sessions = {}

    def _run_agent(
        self, session_id, new_message: types.Content
    ) -> AsyncGenerator[Event, None]:
        return self.runner.run_async(
            session_id=session_id, user_id="self", new_message=new_message
        )

    async def _process_request(
        self,
        new_message: types.Content,
        session_id: str,
        task_updater: TaskUpdater,
    ) -> None:
        session = await self._upsert_session(
            session_id,
        )
        session_id = session.id
        # Run through all events within the request.
        async for event in self._run_agent(session_id, new_message):
            if event.is_final_response():
                parts = convert_genai_parts_to_a2a(event.content.parts)
                logger.debug("✅ Yielding final response: %s", parts)
                await task_updater.add_artifact(parts)
                await task_updater.complete()
                break
            # If the agent is not making a function call, yield an update.
            if not event.get_function_calls():
                logger.debug("⏳ Yielding update response")
                await task_updater.update_status(
                    TaskState.working,
                    message=task_updater.new_agent_message(
                        convert_genai_parts_to_a2a(event.content.parts),
                    ),
                )
            else:
                logger.debug("➡️ Skipping event")

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ):
        # Run the agent until either complete or the task is suspended.
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        # Immediately notify that the task is submitted.
        if not context.current_task:
            updater.submit()
        updater.start_work()
        await self._process_request(
            types.UserContent(
                parts=convert_a2a_parts_to_genai(context.message.parts),
            ),
            context.context_id,
            updater,
        )
        logger.debug("--- 💵💱💶 [Currency] execute exiting ---")

# ... see file for full code

ภายใน currency_agent/__main__.py คือที่ที่คุณเริ่มต้น AgentSkill, AgentCard และสร้างเอเจนต์สกุลเงิน ADK นอกจากนี้ คุณยังตั้งค่าและเริ่มเซิร์ฟเวอร์ A2A ได้ที่นี่ด้วย

A2A Python SDK มีคลาส A2AFastAPIApplication ที่ช่วยให้การเรียกใช้เซิร์ฟเวอร์ HTTP ที่เป็นไปตามข้อกำหนดของ A2A เป็นเรื่องง่าย โดยใช้ FastAPI สำหรับเฟรมเวิร์กเว็บ และโดยทั่วไปจะทำงานร่วมกับเซิร์ฟเวอร์ ASGI เช่น Uvicorn

# ... see file for full code
@click.command()
@click.option("--host", "host", default="localhost")
@click.option("--port", "port", default=10000)
def main(host: str, port: int):
    # Verify one of Google AI Studio or Vertex AI is being used
    if os.getenv("GOOGLE_GENAI_USE_VERTEXAI") != "TRUE" and not os.getenv(
        "GOOGLE_API_KEY"
    ):
        raise ValueError(
            "GOOGLE_API_KEY environment variable not set and "
            "GOOGLE_GENAI_USE_VERTEXAI is not TRUE."
        )

    # A2A Agent Skill definition
    skill = AgentSkill(
        id="get_exchange_rate",
        name="Currency Exchange Rates Tool",
        description="Helps with exchange values between various currencies",
        tags=["currency conversion", "currency exchange"],
        examples=["What is exchange rate between USD and GBP?"],
    )

    # A2A Agent Card definition
    agent_card = AgentCard(
        name="Currency Agent",
        description="Helps with exchange rates for currencies",
        url=f"http://{host}:{port}/",
        version="1.0.0",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
    )

    # Create the ADK runner and executor.
    runner = Runner(
        app_name=agent_card.name,
        agent=root_agent,
        artifact_service=InMemoryArtifactService(),
        session_service=InMemorySessionService(),
        memory_service=InMemoryMemoryService(),
    )
    agent_executor = ADKAgentExecutor(runner, agent_card)

    request_handler = DefaultRequestHandler(
        agent_executor=agent_executor,
        task_store=InMemoryTaskStore(),
    )

    server = A2AFastAPIApplication(
        agent_card=agent_card, http_handler=request_handler
    )

    uvicorn.run(server.build(), host=host, port=port)
# ... see file for full code

หากต้องการเรียกใช้เซิร์ฟเวอร์ A2A ให้เรียกใช้คำสั่งต่อไปนี้ในเทอร์มินัลใหม่

uv run currency_agent/

หากเซิร์ฟเวอร์เริ่มทำงานสำเร็จ เอาต์พุตจะมีลักษณะดังนี้ ซึ่งบ่งชี้ว่าเซิร์ฟเวอร์ทำงานบนพอร์ต 10000

[INFO]: --- 🔧 Loading MCP tools from MCP Server... ---
[INFO]: --- 🤖 Creating ADK Currency Agent... ---
INFO:     Started server process [45824]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:10000 (Press CTRL+C to quit)

ตอนนี้เอเจนต์สกุลเงินทำงานเป็นเซิร์ฟเวอร์ A2A ได้อย่างไม่มีปัญหาแล้ว โดยเอเจนต์หรือไคลเอ็นต์อื่นๆ สามารถเรียกใช้ได้โดยใช้โปรโตคอล A2A

ทดสอบเซิร์ฟเวอร์ A2A

ตอนนี้คุณทดสอบเซิร์ฟเวอร์ได้โดยส่งคำขอไปที่เซิร์ฟเวอร์โดยใช้ A2A

A2A Python SDK มีคลาส a2a.client.A2AClient ที่ช่วยให้คุณทำสิ่งนี้ได้ง่ายขึ้น

ไฟล์ currency_agent/test_client.py มีโค้ดที่เรียกใช้ผ่านกรณีทดสอบต่างๆ หลายกรณีกับเซิร์ฟเวอร์ A2A

# ... see file for full code

# Example test using A2AClient
async def run_single_turn_test(client: A2AClient) -> None:
    """Runs a single-turn non-streaming test."""

    send_message_payload = create_send_message_payload(text="how much is 100 USD in CAD?")
    request = SendMessageRequest(
        id=str(uuid4()), params=MessageSendParams(**send_message_payload)
    )

    print("--- ✉️  Single Turn Request ---")
    # Send Message
    response: SendMessageResponse = await client.send_message(request)
    print_json_response(response, "📥 Single Turn Request Response")
    if not isinstance(response.root, SendMessageSuccessResponse):
        print("received non-success response. Aborting get task ")
        return

    if not isinstance(response.root.result, Task):
        print("received non-task response. Aborting get task ")
        return

    task_id: str = response.root.result.id
    print("--- ❔ Query Task ---")
    # query the task
    get_request = GetTaskRequest(id=str(uuid4()), params=TaskQueryParams(id=task_id))
    get_response: GetTaskResponse = await client.get_task(get_request)
    print_json_response(get_response, "📥 Query Task Response")

# ----- Main Entrypoint (Create client --> Run tests) -----
async def main() -> None:
    """Main function to run the tests."""
    print(f'--- 🔄 Connecting to agent at {AGENT_URL}... ---')
    try:
        async with httpx.AsyncClient() as httpx_client:
            client = await A2AClient.get_client_from_agent_card_url(
                httpx_client, AGENT_URL
            )
            print('--- ✅ Connection successful. ---')

            await run_single_turn_test(client)
            await run_streaming_test(client)
            await run_multi_turn_test(client)

    except Exception as e:
        traceback.print_exc()
        print(f'--- ❌ An error occurred: {e} ---')
        print('Ensure the agent server is running.')

เรียกใช้การทดสอบโดยใช้คำสั่งต่อไปนี้

uv run currency_agent/test_client.py

การทดสอบที่สำเร็จจะส่งผลให้เกิดสิ่งต่อไปนี้

--- 🔄 Connecting to agent at http://localhost:10000... ---
---  Connection successful. ---
--- ✉️ Single Turn Request ---
--- 📥 Single Turn Request Response ---
{"id":"3bc92d7b-d857-4e93-9ff0-b2fb865f6e35","jsonrpc":"2.0","result":{"artifacts":[{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"Based on the current exchange rate, 1 USD is equivalent to 1.3704 CAD. Therefore, 100 USD would be 137.04 CAD.\n"}]}],"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","history":[{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"59819269f7d04849b0bfca7d43ec073c","parts":[{"kind":"text","text":"how much is 100 USD in CAD?"}],"role":"user","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"},{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"286095c6-12c9-40cb-9596-a9676d570dbd","parts":[],"role":"agent","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"}],"id":"52ae2392-84f5-429a-a14b-8413d3d20d97","kind":"task","status":{"state":"completed"}}}

// ...

---  Single Turn Streaming Request ---
---  Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"submitted"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

---  Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

---  Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"message":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"message","messageId":"25f5f972-9475-4e4a-a08d-e13f521d7462","parts":[],"role":"agent","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"},"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

---  Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"artifact":{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"The current exchange rate is 1 EUR to 164.15 JPY. So, 50 EUR would be 8207.5 JPY.\n"}]},"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"artifact-update","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

// ...

--- 🚀 First turn completed, no further input required for this test case. ---

ใช้ได้ผล! คุณได้ทดสอบเรียบร้อยแล้วว่าสามารถสื่อสารกับตัวแทนสกุลเงินผ่านเซิร์ฟเวอร์ A2A ได้ 🎉

ดูกรณีการใช้งานขั้นสูงเพิ่มเติมได้ที่ที่เก็บ a2a-samples ใน GitHub

หากต้องการติดตั้งใช้งานเอเจนต์ Vertex AI Agent Engine มอบประสบการณ์การจัดการสำหรับการติดตั้งใช้งานเอเจนต์ AI ในการใช้งานจริง

9. ขอแสดงความยินดี

ยินดีด้วย คุณสร้างและติดตั้งใช้งานเซิร์ฟเวอร์ MCP ระยะไกล สร้างเอเจนต์สกุลเงินโดยใช้ชุดพัฒนาเอเจนต์ (ADK) ที่เชื่อมต่อกับเครื่องมือโดยใช้ MCP และเปิดเผยเอเจนต์โดยใช้โปรโตคอล Agent2Agent (A2A) เรียบร้อยแล้ว ตอนนี้เอเจนต์สกุลเงินพร้อมโต้ตอบกับเอเจนต์อื่นๆ ในเฟรมเวิร์กใดก็ได้โดยใช้ A2A แล้ว

ที่นี่คือลิงก์ไปยังเอกสารประกอบโค้ดฉบับเต็ม

สิ่งที่เราได้พูดถึงไปแล้ว

  • วิธีสร้างเซิร์ฟเวอร์ MCP ในเครื่อง
  • การทำให้เซิร์ฟเวอร์ MCP ใช้งานได้กับ Cloud Run
  • วิธีสร้างเอเจนต์ด้วย Agent Development Kit ที่ใช้เครื่องมือ MCP
  • วิธีเปิดเผยตัวแทน ADK เป็นเซิร์ฟเวอร์ A2A
  • การทดสอบเซิร์ฟเวอร์ A2A โดยใช้ไคลเอ็นต์ A2A

ล้างข้อมูล

โปรดทำตามขั้นตอนต่อไปนี้เพื่อเลี่ยงไม่ให้เกิดการเรียกเก็บเงินกับบัญชี Google Cloud สำหรับทรัพยากรที่ใช้ในแล็บนี้

  1. ในคอนโซล Google Cloud ให้ไปที่หน้าจัดการทรัพยากร
  2. ในรายการโปรเจ็กต์ ให้เลือกโปรเจ็กต์ที่ต้องการลบ แล้วคลิกลบ
  3. ในกล่องโต้ตอบ ให้พิมพ์รหัสโปรเจ็กต์ แล้วคลิกปิดเพื่อลบโปรเจ็กต์