एमसीपी, एडीके, और A2A का इस्तेमाल शुरू करना

एमसीपी, एडीके, और A2A का इस्तेमाल शुरू करना

इस कोडलैब (कोड बनाना सीखने के लिए ट्यूटोरियल) के बारे में जानकारी

subjectपिछली बार जून 25, 2025 को अपडेट किया गया
account_circleJack Wotherspoon ने लिखा

1. खास जानकारी

एआई एजेंट की लोकप्रियता तेज़ी से बढ़ रही है. ये एजेंट, टास्क को ऑटोमेट करने और फ़ैसले लेने की प्रोसेस में क्रांति ला रहे हैं. ये एजेंट, अपने लक्ष्यों को हासिल करने के लिए, अपने-आप काम करने, सीखने, और अपने आस-पास मौजूद चीज़ों से इंटरैक्ट करने की क्षमता रखते हैं.

लेकिन, एजेंट बनाने का सही तरीका क्या है? इस कोडलैब में, आपको अलग-अलग देशों की मुद्राओं को बदलने वाला करंसी एजेंट बनाने का तरीका बताया गया है. इस सीरीज़ का मकसद, आपको नई टेक्नोलॉजी के बारे में बताना है, ताकि आप उन संक्षिप्त नामों को समझ सकें जिन्हें आपने इंटरनेट पर देखा होगा (एमसीपी, एडीके, A2A).

आर्किटेक्चर

मॉडल कॉन्टेक्स्ट प्रोटोकॉल (एमसीपी)

मॉडल कॉन्टेक्स्ट प्रोटोकॉल (एमसीपी) एक ओपन प्रोटोकॉल है. इससे यह तय होता है कि ऐप्लिकेशन, एलएलएम को कॉन्टेक्स्ट कैसे देते हैं. एमसीपी, एआई मॉडल को संसाधनों, प्रॉम्प्ट, और टूल से कनेक्ट करने का एक स्टैंडर्ड तरीका उपलब्ध कराता है.

एजेंट डेवलपमेंट किट (एडीके)

एजेंट डेवलपमेंट किट (ADK), एआई एजेंट को डेवलप और डिप्लॉय करने के लिए, एक आसान ऑर्केस्ट्रेशन फ़्रेमवर्क है. ADK, मॉडल और डिप्लॉयमेंट के हिसाब से काम करता है. साथ ही, इसे दूसरे फ़्रेमवर्क के साथ काम करने के लिए बनाया गया है. एडीके को एजेंट डेवलपमेंट को सॉफ़्टवेयर डेवलपमेंट की तरह बनाने के लिए डिज़ाइन किया गया था. इससे डेवलपर के लिए, आसान टास्क से लेकर जटिल वर्कफ़्लो तक के एजेंट आर्किटेक्चर को बनाना, डिप्लॉय करना, और ऑर्केस्ट्रेट करना आसान हो जाता है.

Agent2Agent (A2A) प्रोटोकॉल

Agent2Agent (A2A) प्रोटोकॉल एक ओपन स्टैंडर्ड है. इसे एआई एजेंट के बीच आसानी से कम्यूनिकेट करने और साथ मिलकर काम करने के लिए डिज़ाइन किया गया है. जिस तरह एमसीपी, एलएलएम को डेटा और टूल का ऐक्सेस देने के लिए एक स्टैंडर्ड तरीका उपलब्ध कराता है, उसी तरह ए2ए, एजेंटों को एक स्टैंडर्ड तरीका उपलब्ध कराता है, ताकि वे दूसरे एजेंटों से बात कर सकें! आज के समय में, एजेंट अलग-अलग फ़्रेमवर्क का इस्तेमाल करके और अलग-अलग वेंडर के ज़रिए बनाए जाते हैं. ऐसे में, A2A एक सामान्य भाषा उपलब्ध कराता है. इससे अलग-अलग सिस्टम के बीच डेटा शेयर करने में मदद मिलती है.

आपको क्या सीखने को मिलेगा

  • स्थानीय एमसीपी सर्वर बनाने का तरीका
  • एमसीपी सर्वर को Cloud Run पर डिप्लॉय करना
  • एमसीपी टूल का इस्तेमाल करने वाली एजेंट डेवलपमेंट किट की मदद से एजेंट बनाने का तरीका
  • ADK एजेंट को A2A सर्वर के तौर पर एक्सपोज़ करने का तरीका
  • A2A क्लाइंट का इस्तेमाल करके, A2A सर्वर की जांच करना

आपको किन चीज़ों की ज़रूरत होगी

  • कोई ब्राउज़र, जैसे कि Chrome या Firefox
  • बिलिंग की सुविधा वाला Google Cloud प्रोजेक्ट.

2. शुरू करने से पहले

प्रोजेक्ट बनाना

  1. Google Cloud Console में, प्रोजेक्ट चुनने वाले पेज पर, Google Cloud प्रोजेक्ट चुनें या बनाएं.
  2. पक्का करें कि आपके Cloud प्रोजेक्ट के लिए बिलिंग की सुविधा चालू हो. यह देखने का तरीका जानें कि किसी प्रोजेक्ट के लिए बिलिंग की सुविधा चालू है या नहीं.
  3. इस लिंक पर क्लिक करके, Cloud Shell चालू करें. Cloud Shell टर्मिनल (क्लाउड कमांड चलाने के लिए) और एडिटर (प्रोजेक्ट बनाने के लिए) के बीच टॉगल किया जा सकता है. इसके लिए, 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. नीचे दिए गए निर्देश का इस्तेमाल करके, ज़रूरी एपीआई चालू करें. इसमें कुछ मिनट लग सकते हैं.
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 इंस्टॉल करें (इसे डिपेंडेंसी मैनेज करने के लिए इस्तेमाल किया जाता है):
# 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. लोकल एमसीपी सर्वर बनाना

मुद्रा एजेंट को ऑर्केस्ट्रेट करने से पहले, आपको सबसे पहले एमसीपी सर्वर बनाना होगा. इससे, आपको अपने उन टूल को एक्सपोज़ करने में मदद मिलेगी जिनकी आपके एजेंट को ज़रूरत होगी.

एमसीपी सर्वर की मदद से, कम साइज़ वाले प्रोग्राम लिखे जा सकते हैं. इन प्रोग्राम की मदद से, टूल के तौर पर खास सुविधाएं (जैसे, मुद्रा के एक्सचेंज रेट फ़ेच करना) उपलब्ध कराई जा सकती हैं. इसके बाद, कोई एजेंट या एक से ज़्यादा एजेंट, स्टैंडर्ड मॉडल कॉन्टेक्स्ट प्रोटोकॉल (एमसीपी) का इस्तेमाल करके, इन टूल को ऐक्सेस कर सकते हैं.

FastMCP Python पैकेज का इस्तेमाल करके, get_exchange_rate नाम का एक टूल एक्सपोज़ करने वाला एमसीपी सर्वर बनाया जा सकता है. get_exchange_rate टूल, दो मुद्राओं के बीच मौजूदा एक्सचेंज रेट पाने के लिए, इंटरनेट पर Frankfurter API को कॉल करता है.

एमसीपी सर्वर का कोड, 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),
       
)
   
)

एमसीपी सर्वर को स्थानीय तौर पर शुरू करने के लिए, टर्मिनल खोलें और यह कमांड चलाएं. इससे सर्वर http://localhost:8080 पर शुरू हो जाएगा:

uv run mcp-server/server.py

जांचें कि एमसीपी सर्वर सही तरीके से काम कर रहा है या नहीं. साथ ही, यह भी देखें कि मॉडल कॉन्टेक्स्ट प्रोटोकॉल का इस्तेमाल करके, get_exchange_rate टूल को ऐक्सेस किया जा सकता है या नहीं.

नई टर्मिनल विंडो में, यह तरीका अपनाएं, ताकि आप लोकल एमसीपी सर्वर को न रोकें:

uv run mcp-server/test_server.py

आपको 1 डॉलर (अमेरिकन डॉलर) से यूरो (यूरो) के मौजूदा एक्सचेंज रेट का आउटपुट दिखेगा:

--- 🛠️ 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
 
}
} ---

बहुत बढ़िया! आपके पास एक ऐसा एमसीपी सर्वर है जो काम कर रहा है. साथ ही, आपके पास एक ऐसा टूल है जिसे आपका एजेंट ऐक्सेस कर सकता है.

अगले स्टेशन पर जाने से पहले, उस टर्मिनल में Ctrl+C (या Mac पर Command+C) चलाकर, लोकल तौर पर चल रहे एमसीपी सर्वर को बंद करें जहां से आपने इसे शुरू किया था.

5. एमसीपी सर्वर को Cloud Run पर डिप्लॉय करना

अब आपके पास Cloud Run में एमसीपी सर्वर को रिमोट एमसीपी सर्वर के तौर पर डिप्लॉय करने का विकल्प है 🚀☁️

एमसीपी सर्वर को किसी दूसरी जगह से चलाने के फ़ायदे

Cloud Run पर एमसीपी सर्वर को रिमोट तौर पर चलाने से कई फ़ायदे मिल सकते हैं:

  • 📈स्केल करने की सुविधा: Cloud Run को आने वाले सभी अनुरोधों को हैंडल करने के लिए तेज़ी से स्केल करने के लिए बनाया गया है. Cloud Run, मांग के हिसाब से आपके एमसीपी सर्वर को अपने-आप स्केल करेगा.
  • 👥एक ही जगह पर मौजूद सर्वर: IAM की सुविधाओं की मदद से, टीम के सदस्यों के साथ एक ही जगह पर मौजूद एमसीपी सर्वर का ऐक्सेस शेयर किया जा सकता है. इससे वे सभी अपने सर्वर को स्थानीय तौर पर चलाने के बजाय, अपनी लोकल मशीनों से उस सर्वर से कनेक्ट कर सकते हैं. अगर एमसीपी सर्वर में कोई बदलाव किया जाता है, तो टीम के सभी सदस्यों को इसका फ़ायदा मिलेगा.
  • 🔐सुरक्षा: Cloud Run, पुष्टि किए गए अनुरोधों को लागू करने का आसान तरीका उपलब्ध कराता है. इससे आपके एमसीपी सर्वर से सिर्फ़ सुरक्षित कनेक्शन की अनुमति मिलती है. इससे, बिना अनुमति के ऐक्सेस को रोका जा सकता है.

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.

एमसीपी क्लाइंट की पुष्टि करना

आपने पुष्टि करने के लिए --no-allow-unauthenticated को चुना है. इसलिए, रिमोट एमसीपी सर्वर से कनेक्ट करने वाले किसी भी एमसीपी क्लाइंट को पुष्टि करनी होगी.

Cloud Run पर एमसीपी सर्वर होस्ट करने के आधिकारिक दस्तावेज़ों में, इस विषय के बारे में ज़्यादा जानकारी मिलती है. यह जानकारी इस बात पर निर्भर करती है कि एमसीपी क्लाइंट कहां से चलाया जा रहा है.

आपको Cloud Run प्रॉक्सी चलानी होगी, ताकि आपकी लोकल मशीन पर, एमसीपी के रिमोट सर्वर के लिए पुष्टि वाला टनल बनाया जा सके.

डिफ़ॉल्ट रूप से, Cloud Run सेवाओं के यूआरएल के लिए सभी अनुरोधों को Cloud Run Invoker (roles/run.invoker) IAM भूमिका से अनुमति दी जानी चाहिए. आईएएम नीति बाइंडिंग से यह पक्का होता है कि आपके लोकल एमसीपी क्लाइंट की पुष्टि करने के लिए, सुरक्षा के किसी मज़बूत तरीके का इस्तेमाल किया जाए.

आपको यह पक्का करना चाहिए कि आपके या टीम के किसी भी सदस्य के पास, रिमोट एमसीपी सर्वर को ऐक्सेस करने के लिए, अपने आईएएम प्रिंसिपल (Google Cloud खाता) से जुड़ी roles/run.invokerआईएएम भूमिका हो.

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-server/test_server.py फ़ाइल को फिर से चलाएं. इससे यह पक्का किया जा सकेगा कि रिमोट एमसीपी सर्वर काम कर रहा है या नहीं.

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
 
}
} ---

अगर आपको यह पुष्टि करनी है कि रिमोट सर्वर को सचमुच कॉल किया गया था, तो डिप्लॉय किए गए Cloud Run MCP सर्वर के लॉग की क्वेरी की जा सकती है:

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}}

अब आपके पास रिमोट एमसीपी सर्वर है, इसलिए एजेंट बनाने का तरीका जानें! 🤖

6. एजेंट डेवलपमेंट किट (एडीके) की मदद से एजेंट बनाना

आपने एमसीपी सर्वर डिप्लॉय कर लिया है. अब एजेंट डेवलपमेंट किट (एडीके) का इस्तेमाल करके, मुद्रा एजेंट बनाना होगा.

एजेंट डेवलपमेंट किट ने हाल ही में अपना v1.0.0 स्टैबल वर्शन रिलीज़ किया है. इस माइलस्टोन का मतलब है कि Python ADK अब प्रोडक्शन के लिए तैयार है. यह डेवलपर के लिए भरोसेमंद और बेहतर प्लैटफ़ॉर्म है. इसकी मदद से, वे लाइव एनवायरमेंट में अपने एजेंट बना और डिप्लॉय कर सकते हैं.

ADK की मदद से, एजेंट बनाना बेहद आसान हो जाता है. साथ ही, एमसीपी टूल के लिए पहले से मौजूद सहायता की मदद से, उन्हें एमसीपी सर्वर से आसानी से कनेक्ट किया जा सकता है. मुद्रा एजेंट, ADK की MCPToolset क्लास का इस्तेमाल करके, get_exchange_rate टूल को ऐक्सेस करेगा.

मुद्रा एजेंट का कोड 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()

मुद्रा एजेंट की तुरंत जांच करने के लिए, ADK के डेवलपर यूज़र इंटरफ़ेस (यूआई) का फ़ायदा लिया जा सकता है. इसे ऐक्सेस करने के लिए, adk web चलाएं:

uv run adk web

एजेंट को देखने और उसकी जांच करने के लिए, ब्राउज़र में http://localhost:8000 पर जाएं!

पक्का करें कि वेब यूज़र इंटरफ़ेस (यूआई) के सबसे ऊपर बाएं कोने में, एजेंट के तौर पर currency_agent चुना गया हो.

ADK का वेब यूज़र इंटरफ़ेस (यूआई)

चैट बॉक्स में अपने एजेंट से कुछ ऐसा पूछें: "250 कनाडा डॉलर को डॉलर में बदलने पर कितना मिलेगा?". आपको एजेंट को जवाब देने से पहले, हमारे get_exchange_rate एमसीपी टूल को कॉल करते हुए दिखना चाहिए.

ADK वेब मुद्रा एजेंट

एजेंट काम कर रहा है! यह मुद्रा के कन्वर्ज़न 💸 से जुड़ी क्वेरी को हैंडल कर सकता है.

7. Agent2Agent (A2A) प्रोटोकॉल

Agent2Agent (A2A) प्रोटोकॉल एक ओपन स्टैंडर्ड है. इसे एआई एजेंट के बीच आसानी से कम्यूनिकेट करने और साथ मिलकर काम करने के लिए डिज़ाइन किया गया है. इससे अलग-अलग फ़्रेमवर्क का इस्तेमाल करके और अलग-अलग वेंडर के बनाए गए एजेंट, एक ही भाषा में एक-दूसरे से बातचीत कर सकते हैं. इससे अलग-अलग सिस्टम के बीच डेटा शेयर करने में आसानी होती है.

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. करंसी एजेंट A2A सर्वर

अब हम कोड के कुछ हिस्सों पर नज़र डालेंगे और देखेंगे कि A2A सर्वर बनाने वाले अलग-अलग हिस्से एक साथ कैसे काम करते हैं.

फ़ाइल currency_agent/agent_executor.py में जाकर, आपको क्लास ADKAgentExecutor दिखेगी. यह क्लास, A2A ऐब्स्ट्रैक्ट AgentExecutor क्लास से इनहेरिट करती है. यह 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 क्लास उपलब्ध कराता है. इसकी मदद से, A2A के साथ काम करने वाले एचटीटीपी सर्वर को आसानी से चलाया जा सकता है. यह वेब फ़्रेमवर्क के लिए FastAPI का इस्तेमाल करता है. आम तौर पर, इसे Uvicorn जैसे ASGI सर्वर के साथ चलाया जाता है.

# ... 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/

अगर सर्वर सही तरीके से शुरू हो जाता है, तो आउटपुट इस तरह दिखेगा. इससे पता चलता है कि यह पोर्ट 10,000 पर चल रहा है:

[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 सर्वर की मदद से, मुद्रा एजेंट से संपर्क किया जा सकता है! 🎉

इस्तेमाल के ज़्यादा बेहतर उदाहरण देखने के लिए, GitHub पर a2a-samples डेटा स्टोर करने की जगह देखें!

क्या आपको अपना एजेंट डिप्लॉय करना है? Vertex AI Agent Engine, एआई एजेंट को प्रोडक्शन में डिप्लॉय करने के लिए मैनेज किया जा सकने वाला अनुभव देता है!

9. बधाई हो

बधाई हो! आपने रिमोट एमसीपी सर्वर को बनाया और डिप्लॉय किया है. साथ ही, एजेंट डेवलपमेंट किट (एडीके) का इस्तेमाल करके, एमसीपी का इस्तेमाल करने वाले टूल से कनेक्ट होने वाला मुद्रा एजेंट बनाया है. इसके अलावा, Agent2Agent (A2A) प्रोटोकॉल का इस्तेमाल करके, अपने एजेंट को एक्सपोज़ किया है! मुद्रा एजेंट अब A2A का इस्तेमाल करके, किसी भी फ़्रेमवर्क के अन्य एजेंट के साथ इंटरैक्ट करने के लिए उपलब्ध है!

यहां पूरे कोड के दस्तावेज़ का लिंक दिया गया है.

हमने क्या-क्या कवर किया है

  • स्थानीय एमसीपी सर्वर बनाने का तरीका
  • एमसीपी सर्वर को Cloud Run पर डिप्लॉय करना
  • एमसीपी टूल का इस्तेमाल करने वाली एजेंट डेवलपमेंट किट की मदद से एजेंट बनाने का तरीका
  • ADK एजेंट को A2A सर्वर के तौर पर एक्सपोज़ करने का तरीका
  • A2A क्लाइंट का इस्तेमाल करके, A2A सर्वर की जांच करना

व्यवस्थित करें

इस लैब में इस्तेमाल किए गए संसाधनों के लिए, अपने Google Cloud खाते से शुल्क न लिया जाए, इसके लिए यह तरीका अपनाएं:

  1. Google Cloud Console में, संसाधन मैनेज करें पेज पर जाएं.
  2. प्रोजेक्ट की सूची में, वह प्रोजेक्ट चुनें जिसे मिटाना है. इसके बाद, मिटाएं पर क्लिक करें.
  3. डायलॉग बॉक्स में, प्रोजेक्ट आईडी टाइप करें. इसके बाद, प्रोजेक्ट मिटाने के लिए बंद करें पर क्लिक करें.