เกี่ยวกับ Codelab นี้
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
สิ่งที่คุณต้องมี
2 ก่อนเริ่มต้น
สร้างโปรเจ็กต์
- ใน Google Cloud Console ให้เลือกหรือสร้างโปรเจ็กต์ Google Cloud ในหน้าตัวเลือกโปรเจ็กต์
- ตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินสำหรับโปรเจ็กต์ Cloud แล้ว ดูวิธีตรวจสอบว่าได้เปิดใช้การเรียกเก็บเงินในโปรเจ็กต์แล้วหรือไม่
- เปิดใช้งาน Cloud Shell โดยคลิกลิงก์นี้ คุณสลับระหว่างเทอร์มินัล Cloud Shell (สําหรับเรียกใช้คําสั่งคลาวด์) กับ Editor (สําหรับสร้างโปรเจ็กต์) ได้โดยคลิกปุ่มที่เกี่ยวข้องจาก Cloud Shell
- เมื่อเชื่อมต่อกับ Cloud Shell แล้ว ให้ตรวจสอบว่าคุณได้รับการตรวจสอบสิทธิ์แล้วและตั้งค่าโปรเจ็กต์เป็นรหัสโปรเจ็กต์ของคุณโดยใช้คำสั่งต่อไปนี้
gcloud auth list
- เรียกใช้คำสั่งต่อไปนี้ใน Cloud Shell เพื่อยืนยันว่าคำสั่ง gcloud รู้จักโปรเจ็กต์ของคุณ
gcloud config list project
- ใช้คำสั่งต่อไปนี้เพื่อตั้งค่าโปรเจ็กต์
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- เปิดใช้ 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
- ตรวจสอบว่าคุณมี Python 3.10 ขึ้นไป
โปรดดูคำสั่งและการใช้งาน gcloud ในเอกสารประกอบ
3 การติดตั้ง
- โคลนที่เก็บ
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- ติดตั้ง 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"
- กำหนดค่าตัวแปรสภาพแวดล้อม (ผ่านไฟล์
.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
ถามตัวแทนในแชท เช่น "250 CAD เท่ากับกี่ USD" คุณควรเห็นตัวแทนโทรหาเครื่องมือ get_exchange_rate MCP ของเราก่อนที่ตัวแทนจะตอบ
Agent ทำงานแล้ว โดยสามารถจัดการคำค้นหาที่เกี่ยวข้องกับการแปลงสกุลเงิน 💸
7 โปรโตคอล Agent2Agent (A2A)
โปรโตคอล Agent2Agent (A2A) เป็นมาตรฐานแบบเปิดที่ออกแบบมาเพื่อช่วยให้ตัวแทน AI สื่อสารและทำงานร่วมกันได้อย่างราบรื่น ซึ่งช่วยให้เอเจนต์ที่สร้างขึ้นโดยใช้เฟรมเวิร์กที่หลากหลายและโดยผู้ให้บริการที่แตกต่างกันสามารถสื่อสารกันในภาษาเดียวกันได้ ซึ่งจะช่วยลดการทำงานแบบแยกส่วนและส่งเสริมการทำงานร่วมกัน
A2A ช่วยให้ตัวแทนทำสิ่งต่อไปนี้ได้
- ค้นพบ: ค้นหาเอเจนต์อื่นๆ และเรียนรู้ทักษะ (AgentSkill) และความสามารถ (AgentCapabilities) ของเอเจนต์เหล่านั้นโดยใช้การ์ดเอเจนต์ที่ได้มาตรฐาน
- สื่อสาร: แลกเปลี่ยนข้อความและข้อมูลอย่างปลอดภัย
- ทำงานร่วมกัน: มอบหมายงานและประสานงานเพื่อบรรลุเป้าหมายที่ซับซ้อน
โปรโตคอล 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 สำหรับทรัพยากรที่ใช้ในแล็บนี้
- ในคอนโซล Google Cloud ให้ไปที่หน้าจัดการทรัพยากร
- ในรายการโปรเจ็กต์ ให้เลือกโปรเจ็กต์ที่ต้องการลบ แล้วคลิกลบ
- ในกล่องโต้ตอบ ให้พิมพ์รหัสโปรเจ็กต์ แล้วคลิกปิดเพื่อลบโปรเจ็กต์