A2A 에이전트 서비스

이를 통해 애플리케이션은 A2AAgentA2A(에이전트 간) 프로토콜을 통해 노출되는 원격 에이전트에 연결할 수 있습니다. A2A 규격 엔드포인트를 표준 AIAgent으로 래핑하므로 빌드된 프레임워크 또는 기술에 관계없이 익숙한 메서드를 RunAsyncRunStreamingAsync 사용하고 원격 에이전트와 상호 작용할 수 있습니다.

에이전트 프레임워크 에이전트를 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 검색

엔터프라이즈 환경 또는 공용 마켓플레이스에서 에이전트 카드는 종종 중앙 레지스트리에서 관리됩니다. 이러한 레지스트리에서 이미 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 .

비고

선택한 프로토콜 바인딩을 지원하는 엔드포인트에서 원격 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."));

Streaming

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

비고

A2A 에이전트는 스트림의 특정 지점에서 다시 시작을 스트림하는 것이 아니라 스트림 다시 연결을 지원합니다(처음부터 동일한 응답 스트림 가져오기).

Tools

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 는 응답 쿠키를 유지하지 않습니다. 원격 서비스에 인증, 세션 또는 부하 분산 장치 선호도에 쿠키가 필요한 경우 통과http_client=합니다httpx.AsyncClient. 클라이언트 범위를 인증된 보안 주체 하나로 지정하고 애플리케이션에서 닫습니다. 제공된 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)

Streaming

원격 에이전트가 요청을 처리할 때 실시간으로 업데이트를 수신하는 데 사용합니다 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))")

이진 첨부 파일

A2AAgent 는 데이터를 원시 이진 A2A 파트로 보냅니 Content 다. 0-바이트 페이로드는 스트리밍 및 비 스트리밍 실행에서 유효합니다. 미디어 형식, 데이터 URI 매개 변수, base64 인코딩 및 문자 유효성 검사는 여전히 적용됩니다.

장시간 실행 작업

기본적으로 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 저장합니다.

Field	Type	Purpose
`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() 유지하고 A2A 컨텍스트, 작업 ID 및 작업 상태를 함께 유지하여 복원 AgentSession.from_dict(...)합니다.

작업의 TASK_STATE_INPUT_REQUIRED경우 다음 메시지는 동일한 작업을 계속하도록 설정합니다 task_id . 다른 작업 상태의 경우 원격 에이전트가 이전 결과를 구체화하거나 계속할 수 있도록 이전 작업 ID가 전송 reference_task_ids 됩니다.

선택한 실행 컨텍스트 전달

선택 function_invocation_kwargs 하거나 client_kwargs 원격 서비스에 도달해야 하는 경우 허용 목록으로 사용합니다forwarded_kwargs.

async with A2AAgent(
    name="remote",
    url="https://a2a-agent.example.com",
    forwarded_kwargs=["tenant_id", "trace"],
) as agent:
    response = await agent.run(
        "Create the report.",
        function_invocation_kwargs={"tenant_id": "contoso", "secret": "not-forwarded"},
        client_kwargs={"trace": {"id": "abc"}},
    )

나열된 키만 원본 매핑에 따라 그룹화된 아래에 agent_framework전송 SendMessageRequest.metadata 됩니다. 기본값은 실행 kwargs를 보내지 않으며 메타데이터는 HTTP 헤더 또는 메시지 기록에 추가되지 않습니다. 수신 A2A 서비스는 수락할 키와 사용 방법을 결정합니다.

전달된 값은 JSON과 호환되어야 합니다. 매핑 키는 문자열이어야 하고, 숫자는 유한해야 하며, 정수는 포함과 사이 -(2**53)2**53여야 합니다. Protobuf Struct 는 숫자를 double로 저장하므로 정수는 부동 소수점 값으로 도착합니다. 잘못된 선택한 값은 요청을 보내기 전에 발생 AgentInvalidRequestException 합니다.

Authentication

A2A 보안 엔드포인트에 AuthInterceptor를 사용합니다.

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 은(는) 시간(초)을 int 허용합니다 float . 이 값은 연결, 읽기, 쓰기 및 풀 시간 제한 구성 요소에 적용됩니다. 개체를 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:
    ...

시간 제한을 지정하지 않으면 기본값은 10대 연결, 60년대 읽기, 10s 쓰기, 5s 풀입니다.

Tools

Foundry 에이전트가 A2A 에이전트를 도구로 호출하도록 하려면 <a0/>의 팩터리를 참조하세요.

Go는 패키지를 통해 원격 A2A 에이전트를 provider/a2aprovider 지원합니다.

에이전트 프레임워크 및 A2A 패키지를 설치합니다.

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

원격 A2A 에이전트에 연결

원격 에이전트 카드를 해석하여 A2A 클라이언트를 생성하고, 이 클라이언트를 표준 Agent Framework 에이전트로 래핑합니다.

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 및 작업 ID를 저장하므로 후속 메시지는 대화 연속성을 유지할 수 있습니다.

프로토콜 선택

원격 에이전트가 여러 전송 바인딩을 보급하는 경우 A2A 클라이언트를 만들 때 기본 전송을 구성합니다.

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

JSON-RPC를 선호하는 경우 사용합니다 a2a.TransportProtocolJSONRPC .

장기 실행 작업

A2A 작업은 에이전트 프레임워크 연속 토큰을 통해 표시됩니다. 명시적 세션과 agent.AllowBackgroundResponses(true)을 사용해 실행을 시작한 다음, 새 메시지 없이 continuation token을 사용하여 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를 사용하여 에이전트 호스트

더 자세히 살펴보기:

피드백

이 페이지가 도움이 되었나요?

2026-10-16 에 마지막으로 업데이트됨

A2A 에이전트 서비스

시작하기

에이전트 검색

Well-Known URI

Catalog-Based 검색

직접 구성

프로토콜 선택

Streaming

백그라운드 응답

작업 완료를 위한 폴링

스트림 다시 연결

Tools

시작하기

초기화

직접 URL

에이전트 카드

Well-Known URI(A2ACardResolver)

Streaming

이진 첨부 파일

장시간 실행 작업

대화 식별자(context_id)

선택한 실행 컨텍스트 전달

Authentication

시간 제한 구성

Tools

원격 A2A 에이전트에 연결

프로토콜 선택

장기 실행 작업

원격 A2A 에이전트를 도구로 사용

다음 단계

피드백

추가 리소스