إنشاء وكيل أرقام عشوائية متوافق مع بروتوكول A2A: دليل خطوة بخطوة باستخدام بايثون

يُعد بروتوكول التفاعل بين الوكلاء (A2A) معيارًا جديدًا من جوجل يُمكّن وكلاء الذكاء الاصطناعي – بغض النظر عن إطار عملهم الأساسي أو مطورهم – من التواصل والتعاون بسلاسة. يعمل هذا البروتوكول من خلال استخدام رسائل مُعيارية، وبطاقات وكلاء (تصف ما يمكن للوكيل فعله)، والتنفيذ القائم على المهام، مما يسمح للوكلاء بالتفاعل عبر بروتوكول HTTP دون الحاجة إلى منطق تكامل مخصص. يُسهّل A2A بناء أنظمة متعددة الوكلاء قابلة للتطوير وقابلة للتشغيل البيني من خلال تجريد تعقيدات الاتصال. في هذا البرنامج التعليمي، سنقوم بتنفيذ وكيل تجريبي بسيط يُعيد رقمًا عشوائيًا، مما يُساعدك على فهم البنية الأساسية وتدفق بروتوكول A2A من خلال البرمجة العملية.

إعداد التبعيات

سنبدأ أولاً بإعداد بيئتنا ونتابع بتثبيت مدير حزم uv.

على أنظمة ماك أو لينكس:

curl -LsSf https://astral.sh/uv/install.sh | sh

على أنظمة ويندوز (PowerShell):

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

بعد ذلك، سنقوم بإنشاء دليل مشروع جديد وتهيئته باستخدام uv:

uv init a2a-demo
cd a2a-demo

الآن، يمكننا إنشاء وتنشيط بيئة افتراضية:

على أنظمة ماك أو لينكس:

uv venv
source .venv/bin/activate

على أنظمة ويندوز:

uv venv
.venvScriptsactivate

سنقوم الآن بتثبيت التبعيات المطلوبة:

uv add a2a-sdk python-a2a uvicorn

تنفيذ اللبنات الأساسية

منفذ وكيل (agent_executor.py)

في هذه الخطوة، نقوم بتنفيذ المنطق الأساسي لوكيلنا من خلال إنشاء منفذ وكيل، وهو مسؤول عن معالجة الطلبات الواردة وإرجاع الردود بتنسيق A2A. يُغلف RandomNumberAgentExecutor وكيل RandomNumberAgent بسيطًا يُولّد رقمًا عشوائيًا بين 1 و 100. عند ورود طلب، تستدعي طريقة execute منطق الوكيل وتدفع النتيجة إلى قائمة الانتظار للأحداث كرسالة A2A مُعيارية. يشكل هذا الإعداد المنطق الخلفي الذي يمكن لعملاء A2A التفاعل معه.

الكود الكامل متوفر على جيثب

import random
from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.utils import new_agent_text_message
from pydantic import BaseModel

class RandomNumberAgent(BaseModel):
    """يُولّد رقمًا عشوائيًا بين 1 و 100"""
    async def invoke(self) -> str:
        number = random.randint(1, 100)
        return f"تم توليد رقم عشوائي: {number}"

class RandomNumberAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = RandomNumberAgent()
    async def execute(self, context: RequestContext, event_queue: EventQueue):
        result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(result))
    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        raise Exception("إلغاء غير مدعوم")

إعداد خادم A2A وبطاقة الوكيل (main.py)

في هذا القسم، نُعرّف البيانات الوصفية التي تصف ما يمكن لوكيلنا فعله – وهذا ما يُسمى بطاقة الوكيل. فكّر بها كبطاقة عمل الوكيل، التي تحتوي على معلومات مثل اسمه، ووصفه، والمهارات المتاحة، وأنواع الإدخال/الإخراج، والإصدار. نقوم أيضًا بتسجيل مهارات الوكيل، التي تُحدد نوع المهام التي يمكنه التعامل معها. في حالتنا، تتضمن مهارة توليد رقم عشوائي، مُوصّفة بشكل مناسب ومع أمثلة للحث. بمجرد أن تصبح البيانات الوصفية جاهزة، نقوم بتكوين خادم A2A باستخدام A2AStarletteApplication. نوفر بطاقة الوكيل ونربطها بمنطق وكيلنا المخصص باستخدام DefaultRequestHandler، الذي يستخدم RandomNumberAgentExecutor الذي قمنا بتنفيذه سابقًا. أخيرًا، نقوم بتشغيل الخادم باستخدام uvicorn حتى يتمكن الوكيل من البدء في الاستماع إلى رسائل A2A الواردة على المنفذ 9999. يُمكّن هذا الإعداد وكيلنا من استقبال رسائل A2A المُعيارية، ومعالجتها، والرد بطريقة مُنسّقة – باتباع بروتوكول A2A.

الكود الكامل متوفر على جيثب

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from agent_executor import RandomNumberAgentExecutor

def main():
    # تعريف بيانات وصفية للمهارة
    skill = AgentSkill(
        id="random_number",
        name="مُولّد أرقام عشوائية",
        description="يُولّد رقمًا عشوائيًا بين 1 و 100",
        tags=["عشوائي", "رقم", "أداة"],
        examples=["أعطيني رقمًا عشوائيًا", "ألقِ رقمًا", "عشوائي"],
    )
    # تعريف بيانات وصفية للوكيل
    agent_card = AgentCard(
        name="وكيل الأرقام العشوائية",
        description="وكيل يُعيد رقمًا عشوائيًا بين 1 و 100",
        url="http://localhost:9999/",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        skills=[skill],
        version="1.0.0",
        capabilities=AgentCapabilities(),
    )
    # تكوين مُعالِج الطلبات مع منفذ وكيلنا المخصص
    request_handler = DefaultRequestHandler(
        agent_executor=RandomNumberAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )
    # إنشاء خادم تطبيق A2A
    server = A2AStarletteApplication(
        http_handler=request_handler,
        agent_card=agent_card,
    )
    # تشغيل الخادم
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)

if __name__ == "__main__":
    main()

التفاعل مع الوكيل باستخدام A2AClient (client.py)

بعد ذلك، نقوم بإنشاء العميل الذي سيتفاعل مع وكيل A2A الخاص بنا. يقوم برنامج العميل هذا بثلاث مهام رئيسية:

1. جلب بطاقة الوكيل: نبدأ بحل البيانات الوصفية العامة للوكيل باستخدام A2ACardResolver. وهذا يجلب ملف agent.json من نقطة النهاية .well-known/agent.json, والذي يحتوي على تفاصيل أساسية مثل اسم الوكيل، ووصفه، ومهاراته، وقدراته على الاتصال.
2. تهيئة عميل A2A: باستخدام AgentCard المُسترجعة، نقوم بإعداد A2AClient، والذي يُعالِج بروتوكول الاتصال. سيتولى هذا العميل مسؤولية إرسال الرسائل المُنسّقة إلى الوكيل واستقبال الردود.
3. إرسال رسالة واستقبال رد: نقوم بإنشاء رسالة نصية “أعطيني رقمًا عشوائيًا” باستخدام بنية رسالة A2A (Message, Part, TextPart). يتم إرسال الرسالة كجزء من SendMessageRequest، والذي يلفها بمعرف طلب فريد. بمجرد إرسال الرسالة، يُعالِجها الوكيل ويُجيب برقم عشوائي مُولّد، ثم يتم طباعته بتنسيق JSON.

الكود الكامل متوفر على جيثب

import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    Message,
    MessageSendParams,
    Part,
    Role,
    SendMessageRequest,
    TextPart,
)

PUBLIC_AGENT_CARD_PATH = "/.well-known/agent.json"
BASE_URL = "http://localhost:9999"

async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # جلب بطاقة الوكيل
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        try:
            print(f"جلب بطاقة الوكيل العامة من: {BASE_URL}{PUBLIC_AGENT_CARD_PATH}")
            agent_card: AgentCard = await resolver.get_agent_card()
            print("تم جلب بطاقة الوكيل بنجاح:")
            print(agent_card.model_dump_json(indent=2))
        except Exception as e:
            print(f"خطأ في جلب بطاقة الوكيل العامة: {e}")
            return
        # تهيئة عميل A2A مع بطاقة الوكيل
        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
        # بناء الرسالة
        message_payload = Message(
            role=Role.user,
            messageId=str(uuid.uuid4()),
            parts=[Part(root=TextPart(text="أعطيني رقمًا عشوائيًا"))],
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()), params=MessageSendParams(message=message_payload),
        )
        # إرسال الرسالة
        print("إرسال الرسالة...")
        response = await client.send_message(request)
        # طباعة الرد
        print("الرد:")
        print(response.model_dump_json(indent=2))

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

تشغيل الوكيل والاستعلام عنه

لاختبار إعداد A2A الخاص بنا، سنبدأ بتشغيل خادم الوكيل. يتم ذلك بتنفيذ ملف main.py، الذي يُهيئ الوكيل، ويُعرض بطاقة وكيله، ويبدأ في الاستماع إلى الطلبات الواردة على المنفذ 9999.

uv run main.py

بمجرد تشغيل الوكيل، سننتقل إلى برنامج العميل. سيقوم العميل بجلب البيانات الوصفية للوكيل، وإرسال استعلام مُنسّق باستخدام بروتوكول A2A، واستقبال رد. في حالتنا، الاستعلام هو رسالة بسيطة مثل “أعطيني رقمًا عشوائيًا”، وسيُعيد الوكيل رقمًا بين 1 و 100.

uv run client.py

الكود الكامل متوفر على جيثب. جميع الحقوق محفوظة للباحثين في هذا المشروع. تابعونا على تويتر، وانضموا إلى مجتمعنا على ريديت، واشتركوا في قائمتنا البريدية.

المصدر: MarkTechPost