خدمة عامل A2A

A2AAgent يمكن التطبيق الخاص بك من الاتصال بالعوامل البعيدة التي يتم كشفها عبر بروتوكول Agent-to-Agent (A2A). وهو يغلف أي نقطة نهاية متوافقة مع A2A كمعيار AIAgent، حتى تتمكن من استخدام أساليب مألوفة مثل RunAsync والتفاعل RunStreamingAsync مع العوامل البعيدة بغض النظر عن إطار العمل أو التكنولوجيا التي تم إنشاؤها معها.

لعرض عامل إطار عمل عامل كخادم A2A، راجع عوامل المضيف مع A2A.

الشروع في العمل

أضف حزمة NuGet المطلوبة إلى مشروعك:

dotnet add package Microsoft.Agents.AI.A2A --prerelease

اكتشاف العامل

قبل الاتصال بعامل A2A عن بعد، تحتاج إلى اكتشافه وإنشاء مثيل AIAgent . يحدد بروتوكول A2A ثلاث استراتيجيات اكتشاف، كل منها مدعوم من إطار عمل العامل.

Well-Known URI

يمكن لوكلاء A2A جعل بطاقة الوكيل الخاصة بهم قابلة للاكتشاف في مسار موحد: https://{domain}/.well-known/agent-card.json. A2ACardResolver استخدم لجلب البطاقة وإنشاء وكيل في مكالمة واحدة:

using A2A;
using Microsoft.Agents.AI;

// Initialize a resolver pointing at the remote agent's host.
A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));

// Resolve the agent card and create an AIAgent in one step.
AIAgent agent = await resolver.GetAIAgentAsync();

// Use the agent.
Console.WriteLine(await agent.RunAsync("Hello!"));

Tip

GetAIAgentAsync يقبل أيضا معلمة اختيارية A2AClientOptionsلتحديد البروتوكول.

Catalog-Based Discovery

في بيئات المؤسسة أو الأسواق العامة، غالبا ما تتم إدارة بطاقات الوكيل بواسطة سجل مركزي. إذا كان لديك بالفعل تم AgentCard الحصول عليه من مثل هذا السجل، فحوله مباشرة إلى AIAgent:

using A2A;
using Microsoft.Agents.AI;

// Assume agentCard was retrieved from a registry or catalog.
AgentCard agentCard = await GetAgentCardFromRegistryAsync("travel-planner");

AIAgent agent = agentCard.AsAIAgent();

Console.WriteLine(await agent.RunAsync("Plan a trip to Paris."));

التكوين المباشر

بالنسبة للأنظمة المقترنة بإحكام أو سيناريوهات التطوير حيث تكون نقطة نهاية العامل معروفة مسبقا، قم بإنشاء A2AClient مباشرة وتحويلها إلى AIAgent:

using A2A;
using Microsoft.Agents.AI;

// Create a client pointing at the known agent endpoint.
A2AClient a2aClient = new(new Uri("https://a2a-agent.example.com"));

AIAgent agent = a2aClient.AsAIAgent(name: "my-agent", description: "A helpful assistant.");

Console.WriteLine(await agent.RunAsync("What can you help me with?"));

تحديد البروتوكول

يمكن أن يعرض وكلاء A2A روابط بروتوكول متعددة مثل HTTP+JSON وJSON-RPC. بشكل افتراضي، يفضل HTTP+JSON على JSON-RPC. استخدم A2AClientOptions.PreferredBindings للتحكم بشكل صريح في ربط البروتوكول المستخدم:

Note

يجب أن يكون عامل A2A البعيد متوفرا في نقطة نهاية تدعم ربط البروتوكول المحدد.

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver agentCardResolver = new(new Uri("https://a2a-agent.example.com"));

AgentCard agentCard = await agentCardResolver.GetAgentCardAsync();

// Prefer HTTP+JSON protocol binding. For JSON-RPC, set PreferredBindings = [ProtocolBindingNames.JsonRpc]
A2AClientOptions options = new()
{
    PreferredBindings = [ProtocolBindingNames.HttpJson]
};

AIAgent agent = agentCard.AsAIAgent(options: options);

Console.WriteLine(await agent.RunAsync("Tell me a joke about a pirate."));

البث المباشر

يدعم A2A استجابات الدفق عبر أحداث Server-Sent. يستخدم RunStreamingAsync لتلقي التحديثات في الوقت الفعلي حيث يعالج العامل البعيد الطلب:

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

await foreach (var update in agent.RunStreamingAsync("Write a short story about a robot."))
{
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

استجابات الخلفية

يدعم وكلاء A2A استجابات الخلفية للتعامل مع العمليات طويلة الأمد. عندما يقوم عامل A2A بعيد بإرجاع مهمة بدلا من رسالة فورية، يوفر إطار عمل العامل رمز متابعة يمكنك استخدامه للاستقصاء عن النتائج أو إعادة الاتصال بالتدفقات المتقطعة.

التحقق من اكتمال المهمة

بالنسبة للسيناريوهات غير المتدفقة، استخدم AllowBackgroundResponses لتلقي رمز متابعة واستقصاء حتى تكتمل المهمة:

using A2A;
using Microsoft.Agents.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

AgentSession session = await agent.CreateSessionAsync();

// AllowBackgroundResponses must be true so the server returns immediately with a continuation token
// instead of blocking until the task is complete.
AgentRunOptions options = new() { AllowBackgroundResponses = true };

// Start the initial run with a long-running task.
AgentResponse response = await agent.RunAsync(
    "Conduct a comprehensive analysis of quantum computing applications in cryptography.",
    session,
    options: options);

// Poll until the response is complete.
while (response.ContinuationToken is { } token)
{
    // Wait before polling again.
    await Task.Delay(TimeSpan.FromSeconds(2));

    // Continue with the token.
    response = await agent.RunAsync(session, options: new AgentRunOptions { ContinuationToken = token });
}

Console.WriteLine(response);

إعادة اتصال الدفق

في سيناريوهات الدفق، قد يتضمن كل تحديث رمزا مميزا للمتابعة. إذا تمت مقاطعة الدفق، فاستخدم الرمز المميز لإعادة الاتصال والحصول على تدفق الاستجابة من البداية:

using A2A;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

A2ACardResolver resolver = new(new Uri("https://a2a-agent.example.com"));
AIAgent agent = await resolver.GetAIAgentAsync();

AgentSession session = await agent.CreateSessionAsync();

ResponseContinuationToken? continuationToken = null;

await foreach (var update in agent.RunStreamingAsync(
    "Conduct a comprehensive analysis of quantum computing applications in cryptography.",
    session))
{
    // Save the continuation token to reconnect later if the stream is interrupted.
    // Continuation tokens are only returned for long-running tasks. If the A2A agent
    // returns a message instead of a task, the continuation token will not be initialized.
    if (update.ContinuationToken is { } token)
    {
        continuationToken = token;
    }
}

// If the stream was interrupted and a continuation token was captured,
// reconnect to the response stream using the saved continuation token.
if (continuationToken is not null)
{
    await foreach (var update in agent.RunStreamingAsync(
        session,
        options: new() { ContinuationToken = continuationToken }))
    {
        if (!string.IsNullOrEmpty(update.Text))
        {
            Console.WriteLine(update.Text);
        }
    }
}

Note

يدعم وكلاء A2A إعادة اتصال الدفق (الحصول على نفس تدفق الاستجابة من البداية)، وليس استئناف الدفق من نقطة معينة في الدفق.

ادوات

A2AAgent هو برنامج تضمين على مستوى النقل حول عامل A2A بعيد. أيا كانت الأدوات التي يستخدمها العامل البعيد مباشرة على الجانب البعيد وغير مرئية للتعليمات البرمجية الخاصة بك. لا يتم تكوين أنواع أدوات إطار عمل العامل (أدوات الوظيفة، ومترجم التعليمات البرمجية، والبحث في الملفات، وMCP المستضافة/المحلية، وما إلى ذلك) على A2AAgent نفسها — لتوسيع قدرات العامل البعيد، وتغيير تكوين العامل البعيد.

الشروع في العمل

تثبيت حزمة A2A:

pip install agent-framework-a2a --pre

التهيئة

A2AAgent يمكن تهيئتها بثلاث طرق اعتمادا على مقدار ما تعرفه عن العامل البعيد في وقت متقدم.

عنوان URL المباشر

للتطوير أو الأنظمة المقترنة بإحكام حيث تكون نقطة النهاية معروفة:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    response = await agent.run("Hello!")
    print(response.messages[0].text)

عند توفير عنوان URL فقط، A2AAgent ينشئ الحد الأدنى من بطاقة العامل داخليا ويتصل باستخدام JSON-RPC.

عميل HTTP الذي A2AAgent يقوم بإنشائه لا يستمر في ملفات تعريف الارتباط للاستجابة. إذا كانت الخدمة البعيدة تتطلب ملفات تعريف الارتباط للمصادقة أو الجلسات أو ترابط موازن التحميل، فمرر httpx.AsyncClient من خلال http_client=. حدد نطاق العميل إلى كيان واحد مصادق عليه وأغلقه في التطبيق الخاص بك. يظل عملاء HTTP المتوفرون مملوكين للمتصل، بما في ذلك عند اجتيازك client=أيضا .

بطاقة الوكيل

إذا كان لديك سجل AgentCard أو كتالوج، فمرره مباشرة:

from agent_framework.a2a import A2AAgent

async with A2AAgent(agent_card=agent_card) as agent:
    response = await agent.run("Plan a trip to Paris.")
    print(response.messages[0].text)

AgentCard عند توفير، A2AAgent يتم تعيين الإعدادات الافتراضية name ومن description البطاقة. وهو يتفاوض على النقل باستخدام بطاقة supported_interfaces.

Well-Known URI (A2ACardResolver)

استخدم A2ACardResolver من a2a-sdk لاكتشاف العامل البعيد في المسار القياسي المعروف (/.well-known/agent.json):

import httpx
from a2a.client import A2ACardResolver
from agent_framework.a2a import A2AAgent

async with httpx.AsyncClient(timeout=60.0) as http_client:
    resolver = A2ACardResolver(httpx_client=http_client, base_url="https://a2a-agent.example.com")
    agent_card = await resolver.get_agent_card()

async with A2AAgent(agent_card=agent_card) as agent:
    response = await agent.run("What can you help me with?")
    print(response.messages[0].text)

البث المباشر

يستخدم stream=True لتلقي التحديثات في الوقت الفعلي حيث يعالج العامل البعيد الطلب:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    stream = agent.run("Write a short story about a robot.", stream=True)
    async for update in stream:
        for content in update.contents:
            if content.text:
                print(content.text, end="", flush=True)

    final = await stream.get_final_response()
    print(f"\n({len(final.messages)} message(s))")

مهام Long-Running

بشكل افتراضي، A2AAgent ينتظر حتى ينتهي العامل البعيد قبل العودة. بالنسبة للمهام طويلة الأمد، قم بتعيين background=True إلى عرض رمز متابعة مميز يمكنك استخدامه للاستقصاء أو الاشتراك لاحقا:

from agent_framework.a2a import A2AAgent

async with A2AAgent(name="worker", url="https://a2a-agent.example.com") as agent:
    # Start a long-running task
    response = await agent.run("Process this large dataset", background=True)

    if response.continuation_token:
        # Poll for completion later
        result = await agent.poll_task(response.continuation_token)
        print(result)

يمكنك أيضا إعادة الاشتراك في دفق SSE بدلا من الاستقصاء:

# Resubscribe to the task's event stream
response = await agent.run(continuation_token=response.continuation_token)

هوية المحادثة (context_id)

A2AAgent يخزن حالة البروتوكول الدائم في AgentSession.service_session_id كرسم A2AServiceSessionId خرائط:

الحقل	Type	الغرض
context_id	str	تعريف محادثة A2A.
task_id	str \| None	يتعقب أحدث مهمة بعيدة، عندما أنشأت الاستجابة واحدة.
task_state	TaskState \| None	يسجل حالة المهمة الأخيرة حتى يتمكن الطلب التالي من متابعة مهمة مطلوبة للإدخال أو الرجوع إلى مهمة مكتملة.

إنشاء جلسة عمل بحالة منظمة عندما يعرف التطبيق الخاص بك بالفعل سياق A2A:

from agent_framework import AgentSession
from agent_framework.a2a import A2AAgent, A2AServiceSessionId

async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
    session = AgentSession(
        service_session_id=A2AServiceSessionId(
            context_id="my-conversation-1",
            task_id=None,
            task_state=None,
        )
    )

    # The A2A message uses context_id="my-conversation-1".
    response = await agent.run("Hello!", session=session)

    # A2AAgent updates task_id and task_state from the response.
    response = await agent.run("Follow-up question", session=session)

يمكنك أيضا البدء ب AgentSession() والسماح A2AAgent بملء التعيين المنظم من الاستجابة الأولى. استمر في جلسة العمل العادية مع session.to_dict() واستعادتها باستخدام AgentSession.from_dict(...)؛ يظل سياق A2A ومعرف المهمة وحالة المهمة معا.

بالنسبة لمهمة في TASK_STATE_INPUT_REQUIRED، تعين الرسالة التالية ذلك task_id لمتابعة نفس المهمة. بالنسبة إلى حالات المهام الأخرى، يتم إرسال معرف المهمة السابق من خلال reference_task_ids بحيث يمكن للعامل البعيد تحسين النتيجة السابقة أو المتابعة منها.

Authentication

AuthInterceptor استخدم لنقاط نهاية A2A الآمنة:

from a2a.client.auth.interceptor import AuthInterceptor
from agent_framework.a2a import A2AAgent

class BearerAuth(AuthInterceptor):
    def __init__(self, token: str):
        self.token = token

    async def intercept(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        return request

async with A2AAgent(
    name="secure-agent",
    url="https://secure-a2a-agent.example.com",
    auth_interceptor=BearerAuth("your-token"),
) as agent:
    response = await agent.run("Hello!")

تكوين المهلة

A2AAgent يقبل أو intfloat عدد من الثوان. تنطبق القيمة على مكونات مهلة الاتصال والقراءة والكتابة والتجمع. مرر كائن httpx.Timeout لتكوين هذه القيم بشكل منفصل:

import httpx
from agent_framework.a2a import A2AAgent

# Simple timeout (applies to all components)
async with A2AAgent(name="remote", url="https://example.com", timeout=120) as agent:
    ...

# Fine-grained timeout
async with A2AAgent(
    name="remote",
    url="https://example.com",
    timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=5.0),
) as agent:
    ...

عندما لا يتم تحديد مهلة، تكون الإعدادات الافتراضية هي: اتصال 10s، قراءة 60s، كتابة 10s، تجمع 5s.

ادوات

A2AAgent هو برنامج تضمين على مستوى النقل حول عامل A2A بعيد. أيا كانت الأدوات التي يستخدمها العامل البعيد مباشرة على الجانب البعيد وغير مرئية للتعليمات البرمجية الخاصة بك. لا يتم تكوين أنواع أدوات إطار عمل العامل (أدوات الوظيفة، ومترجم التعليمات البرمجية، والبحث في الملفات، وMCP المستضافة/المحلية، وما إلى ذلك) على A2AAgent نفسها — لتوسيع قدرات العامل البعيد، وتغيير تكوين العامل البعيد.

إذا كنت تريد أن يتصل عامل Foundry بعامل A2A كأداة، فشاهد المصنع علىget_a2a_toolFoundryChatClient.

يدعم Go وكلاء A2A عن بعد من خلال الحزمة provider/a2aprovider .

تثبيت حزم إطار عمل العامل وA2A:

go get github.com/microsoft/agent-framework-go
go get github.com/a2aproject/a2a-go/v2

الاتصال بعامل A2A بعيد

حل بطاقة العامل البعيد، وإنشاء عميل A2A منها، والتفاف العميل كعامل إطار عمل عامل قياسي:

import (
    "context"

    "github.com/a2aproject/a2a-go/v2/a2aclient"
    "github.com/a2aproject/a2a-go/v2/a2aclient/agentcard"
    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/provider/a2aprovider"
)

ctx := context.Background()

card, err := agentcard.DefaultResolver.Resolve(ctx, "http://localhost:5000")
if err != nil {
    panic(err)
}

client, err := a2aclient.NewFromCard(ctx, card)
if err != nil {
    panic(err)
}

a := a2aprovider.NewAgent(
    client,
    a2aprovider.AgentConfig{
        Config: agent.Config{
            Name:        card.Name,
            Description: card.Description,
        },
    },
)

resp, err := a.RunText(ctx, "Hello!").Collect()

يقوم الموفر بتخزين A2A context_id ومعرفات المهام في جلسة عمل إطار العامل بحيث يمكن لرسائل المتابعة الحفاظ على استمرارية المحادثة.

تحديد البروتوكول

إذا أعلن عامل بعيد عن روابط نقل متعددة، فكون النقل المفضل عند إنشاء عميل A2A:

client, err := a2aclient.NewFromCard(
    ctx,
    card,
    a2aclient.WithConfig(a2aclient.Config{
        PreferredTransports: []a2a.TransportProtocol{a2a.TransportProtocolHTTPJSON},
    }),
)

استخدم a2a.TransportProtocolJSONRPC عندما تريد تفضيل JSON-RPC.

المهام طويلة الأمد

تظهر مهام A2A من خلال رموز متابعة إطار العامل. ابدأ التشغيل بجلسة عمل صريحة و agent.AllowBackgroundResponses(true)، ثم الاستقصاء عن طريق استدعاء Run بدون رسائل جديدة ورمز المتابعة المميز:

session, err := a.CreateSession(ctx)
if err != nil {
    panic(err)
}

resp, err := a.RunText(
    ctx,
    "Process this large dataset.",
    agent.WithSession(session),
    agent.AllowBackgroundResponses(true),
).Collect()
if err != nil {
    panic(err)
}

for resp.ContinuationToken != "" {
    resp, err = a.Run(
        ctx,
        nil,
        agent.WithSession(session),
        agent.WithContinuationToken(resp.ContinuationToken),
    ).Collect()
    if err != nil {
        panic(err)
    }
}

بالنسبة إلى عمليات تشغيل الدفق المتقطعة، قم بالتقاط update.ContinuationToken آخر تحديث تم تلقيه وتمريره إلى تشغيل دفق لاحق باستخدام agent.WithContinuationToken(token) و agent.Stream(true).

استخدام عوامل A2A عن بعد كأدوات

حل كل عامل بعيد، والتفافه مع a2aprovider.NewAgent، وتحويله إلى أداة باستخدام agenttool.New.

tools := make([]tool.Tool, 0, len(agentURLs))

for _, agentURL := range agentURLs {
    card, err := agentcard.DefaultResolver.Resolve(ctx, agentURL)
    if err != nil {
        panic(err)
    }

    client, err := a2aclient.NewFromCard(ctx, card)
    if err != nil {
        panic(err)
    }

    remoteAgent := a2aprovider.NewAgent(client, a2aprovider.AgentConfig{
        Config: agent.Config{
            Name:        card.Name,
            Description: card.Description,
        },
    })

    tools = append(tools, agenttool.New(remoteAgent, agenttool.Config{}))
}

Tip

راجع عينة موفر A2Aوعوامل A2A كعينة أدوات للحصول على أمثلة كاملة قابلة للتشغيل.

الخطوات التالية

وكلاء المضيف مع A2A

انتقل إلى أبعد من ذلك:

• وكلاء الجمارك
• مواصفة بروتوكول A2A