.NET A2A 호스팅 패키지를 사용하여 ASP.NET Core 통해 Agent Framework 에이전트를 노출합니다. 전체 다 언어 서버 가이드 는 A2A가 있는 호스트 에이전트 를 참조하세요.
공식 A2A Go provider/a2aprovider 서버 처리기와 함께 Go 패키지를 사용합니다. 전체 서버 예제 는 A2A가 있는 호스트 에이전트 를 참조하세요.
Agent Framework는 공식 A2A SDK를 통해 에이전트 및 워크플로를 호스팅하기 위한 두 가지 Python 패키지를 제공합니다.
| Package | 통합 모델 | 사용해야 하는 경우 |
|---|---|---|
agent-framework-a2a |
요청을 변환하고, 에이전트를 실행하고, A2A 작업 이벤트 및 아티팩트를 게시하는 의견 A2AExecutor 입니다. |
표준 에이전트 프레임워크-A2A 동작을 원하고 A2A SDK 서버를 어셈블하기만 하면 됩니다. |
agent-framework-hosting-a2a |
앱에서 소유하는 실행기를 위한 점진적 기본 구성 요소입니다. 기본 에이전트 또는 워크플로 변환기부터 시작하고, 필요에 따라 이러한 변환기를 기반으로 네이티브 카드 생성과 모드 유효성 검사를 추가하는 AgentA2AAdapter 또는 WorkflowA2AAdapter을 사용할 수 있습니다. |
애플리케이션은 세션 매핑, 작업 전환, 이벤트 전달, 아티팩트 경계, 출력 변환 또는 다중 프로토콜 호스트를 소유해야 합니다. |
두 패키지 모두 네이티브 A2A SDK 형식 및 서버 구성 요소를 사용합니다. 애플리케이션은 요청 처리기, 작업 저장소, 경로 또는 SDK 애플리케이션 작성기, 인증 및 배포를 제공합니다. 이 agent-framework-hosting-a2a경우 애플리케이션은 에이전트 카드를 직접 생성하거나 어댑터에서 에이전트 카드를 생성하도록 할 수 있습니다.
의견 있는 A2A 실행기 사용
기본 제공 서버 어댑터가 라이프사이클과 일치하는 경우 agent-framework-a2a를 설치하세요:
pip install --pre agent-framework-a2a starlette uvicorn
A2AExecutor 는 A2A SDK를 AgentExecutor구현합니다. A2A 요청 컨텍스트에서 사용자 입력을 읽고, A2A 컨텍스트 ID에서 에이전트 프레임워크 세션을 만들고, 스트리밍 또는 비 스트리밍 모드에서 에이전트를 실행하고, 지원되는 출력 콘텐츠를 변환하고, SDK를 TaskUpdater통해 작업 상태 및 아티팩트 이벤트를 게시합니다.
A2A SDK DefaultRequestHandler, 작업 저장소, 에이전트 카드 및 Starlette 애플리케이션 또는 지원되는 다른 서버 통합을 사용하여 작성합니다.
A2AExecutor(agent, stream=True)로 스트리밍을 구성하고, 안정적인 에이전트 실행 옵션을 run_kwargs을 통해 전달하거나, 다른 출력 매핑이 필요할 경우 A2AExecutor를 서브클래싱하고 handle_events를 재정의합니다.
A2AExecutor 는 A2A 엔드포인트로 범위가 지정되고 A2A 실행 및 세션 매핑을 직접 관리합니다. 한 애플리케이션에서 여러 프로토콜을 통해 동일한 에이전트를 사용할 수 있어야 하는 경우 호스팅 패키지를 사용합니다.
전체 서버 설정은 A2A를 통해 에이전트 프레임워크 에이전트 노출을 참조하세요.
앱 소유 실행기에서 어댑터 사용
애플리케이션이 네이티브 A2A 실행기를 소유하지만 에이전트 프레임워크가 공용 카드를 생성하고 변환의 유효성을 검사하려고 할 때 호스팅 패키지를 설치합니다.
pip install --pre agent-framework-hosting-a2a starlette uvicorn
AgentA2AAdapter 에이전트 또는 AgentState을 수락합니다. 비동 get_card 기 메서드는 공용 이름 및 설명을 파생시키고, 기본적으로 보수적인 텍스트 모드를 사용하며, 에이전트 프레임워크 SkillsProvider 인스턴스에서 네이티브 A2A 기술을 유추할 수 있습니다. 서버 기능 및 지원되는 인터페이스는 에이전트 run 의 메서드가 아닌 애플리케이션 엔드포인트를 설명하기 때문에 명시적으로 유지됩니다.
어댑터는 기본적으로 구성된 카드 모드에 따라 값의 유효성을 검사하는 a2a_to_run 및 a2a_from_run 메서드를 노출합니다. 애플리케이션은 여전히 A2A 실행기, 작업 수명 주기, 이벤트 큐, 아티팩트 경계, 세션 정책, 인증, 경로 및 배포를 소유합니다.
이 실행기는 인바운드 변환, 에이전트 상태 및 아웃바운드 변환에 하나의 어댑터를 사용합니다.
class AppAgentExecutor(AgentExecutor):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, adapter: AgentA2AAdapter[Any]) -> None:
self.adapter = adapter
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = self.adapter.a2a_to_run(context.message, stream=True)
agent = await self.adapter.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.adapter.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = self.adapter.a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = self.adapter.a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.adapter.state.set_session(session_id, session)
await updater.complete()
except asyncio.CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
서버 설정은 어댑터를 만들고, 네이티브 AgentCard를 생성하고, A2A SDK 요청 처리기를 사용하여 앱 소유 실행기를 구성합니다.
if __name__ == "__main__":
flight_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="flight-booking",
description="Search and book flights across Europe.",
),
instructions="Help users search and book flights across Europe.",
)
hotel_skill = InlineSkill(
frontmatter=SkillFrontmatter(
name="hotel-booking",
description="Search and book hotels across Europe.",
),
instructions="Help users search and book hotels across Europe.",
)
agent = Agent(
client=OpenAIChatClient(),
name="Europe Travel Agent",
description="Helps users search and book flights and hotels across Europe.",
instructions="You are a helpful Europe Travel Agent.",
context_providers=[SkillsProvider([flight_skill, hotel_skill])],
)
state = AgentState(agent)
adapter = AgentA2AAdapter(
state,
version="1.0.0",
capabilities=AgentCapabilities(streaming=True),
supported_interfaces=[AgentInterface(url="http://localhost:9999/", protocol_binding="JSONRPC")],
)
public_agent_card = asyncio.run(adapter.get_card())
request_handler = DefaultRequestHandler(
agent_executor=AppAgentExecutor(adapter),
task_store=InMemoryTaskStore(),
agent_card=public_agent_card,
)
앱 소유의 A2A 실행기 구축
애플리케이션에서 카드 생성을 직접 제어해야 하는 경우 독립 실행형 호스팅 도우미를 사용합니다.
pip install --pre agent-framework-hosting-a2a starlette uvicorn
도우미는 프레임워크 중립적입니다.
-
a2a_to_run는 A2AMessage를 Agent Framework 실행 인수로 변환합니다. -
a2a_from_run는 에이전트 프레임워크 응답 및 스트리밍 업데이트를 A2APart값으로 변환합니다.
실행기는 세션 키를 선택하고 작업 전환, 이벤트 큐, 아티팩트 ID, 메시지 경계 및 아웃바운드 배달을 소유합니다.
a2a_from_run 는 애플리케이션이 해당 부분을 A2A 메시지 또는 아티팩트에 그룹화하고 메시지 수준 메타데이터를 적용할 수 있도록 플랫 파트 목록을 반환합니다.
호스팅 설정은 다중 프로토콜 애플리케이션도 지원합니다. A2A, OpenAI Responses, Telegram 및 MCP 경로 전반에서 동일한 에이전트 대상과 AgentState 인프라를 공유하면서도, 각 프로토콜 엔드포인트는 자체 변환, 권한 부여 및 세션 키 정책을 유지합니다. 이렇게 하면 클라이언트가 각 엔드포인트에 대해 별도의 에이전트 배포를 만들지 않고 동시에 서로 다른 프로토콜을 통해 하나의 에이전트에 연결할 수 있습니다.
네이티브 A2A SDK 실행기에서 도우미를 작성합니다. 이 샘플은 A2A 작업을 만들고 업데이트하고, 인바운드 메시지를 에이전트 프레임워크 실행으로 변환하고, 스트림이 완료된 후 업데이트된 AgentState 세션을 유지하며, 반환된 파트를 아티팩트로 게시합니다.
class AppAgentExecutor(AgentExecutor, Generic[AgentT]):
"""Native A2A SDK executor composed with Agent Framework conversion helpers."""
def __init__(self, state: AgentState[AgentT]) -> None:
self.state = state
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.context_id is None:
raise ValueError("A2A context id is required")
updater = TaskUpdater(event_queue, context.task_id or "", context.context_id)
await updater.cancel()
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.message is None or context.context_id is None:
raise ValueError("A2A message and context id are required")
task = context.current_task
if task is None:
task = new_task_from_user_message(context.message)
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, context.context_id)
await updater.submit()
try:
await updater.start_work()
run = a2a_to_run(context.message, stream=True)
agent = await self.state.get_target()
# Demo-only key: the outer server must authenticate and authorize these protocol IDs for multi-user use.
session_id = f"a2a:{context.tenant}:{context.context_id}"
session = await self.state.get_or_create_session(session_id)
if not run["stream"]:
raise RuntimeError("This executor requires streaming run arguments.")
stream = agent.run( # pyright: ignore[reportCallIssue]
run["messages"],
session=session,
options=run["options"],
stream=run["stream"],
)
default_artifact_id = uuid.uuid4().hex
streamed_artifact_ids: set[str] = set()
async for update in stream:
parts = a2a_from_run(update)
if parts:
artifact_id = update.message_id or default_artifact_id
await updater.add_artifact(
parts=parts,
artifact_id=artifact_id,
append=True if artifact_id in streamed_artifact_ids else None,
)
streamed_artifact_ids.add(artifact_id)
final_response = await stream.get_final_response()
if not streamed_artifact_ids:
parts = a2a_from_run(final_response)
if parts:
await updater.update_status(
state=TaskState.TASK_STATE_WORKING,
message=updater.new_agent_message(parts),
)
await self.state.set_session(session_id, session)
await updater.complete()
except CancelledError:
await updater.update_status(state=TaskState.TASK_STATE_CANCELED)
except Exception:
logger.exception("A2A agent execution failed.")
await updater.update_status(
state=TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text="Agent execution failed.")]),
)
이 샘플에서는 Starlette와 Uvicorn을 사용하지만 도우미는 둘 중 하나에 연결되지 않습니다. 애플리케이션 프레임워크 또는 A2A SDK 애플리케이션 작성기를 사용하여 A2A 에이전트 카드 및 JSON-RPC 경로를 제공합니다.
# Create the Agent Framework agent for the chosen type
agent_factory = AGENT_FACTORIES[args.agent_type]
agent = agent_factory(client)
state = AgentState(agent)
# Build the A2A server components
url = f"http://{args.host}:{args.port}/"
agent_card = AGENT_CARD_FACTORIES[args.agent_type](url)
executor = AppAgentExecutor(state)
task_store = InMemoryTaskStore()
request_handler = DefaultRequestHandler(
agent_executor=executor,
task_store=task_store,
agent_card=agent_card,
)
app = Starlette(
routes=[
*create_agent_card_routes(agent_card),
*create_jsonrpc_routes(request_handler, "/"),
]
)
어댑터를 사용하여 워크플로 호스트
WorkflowA2AAdapter 는 워크플로 또는 WorkflowState에 대해 동일한 카드 생성 및 변환 경계를 제공합니다. 워크플로의 선언된 형식에서 보수적인 입력 및 출력 모드를 유추하거나 애플리케이션별 표현에 대한 명시적 모드를 제공할 수 있습니다.
독립 실행형 a2a_to_workflow_run 및 a2a_from_workflow_run 도우미는 형식화된 워크플로 입력 및 출력 변환을 제공합니다. 어댑터는 적용되는 카드 모드에 따라 유효성을 검사하는 비동기 a2a_to_run 및 동기 a2a_from_run 메서드로 이를 제공합니다. 입력 변환은 워크플로의 단일 시작 실행기 입력 형식에 대해 하나의 A2A 텍스트, 원시 또는 데이터 파트를 허용하고 출력 변환 맵은 공용 워크플로 출력을 네이티브 A2A 파트로 완료했습니다. 어댑터가 출력 모드를 유추해야 하는 경우 유효성이 검사된 출력 변환 전에 호출 get_card 합니다.
애플리케이션은 네이티브 A2A 실행기와 진행 상황, 작업 상태, 아티팩트, 체크포인트 및 human-in-the-loop 후속 진행의 스트리밍에 대한 책임을 계속 집니다. 보류 중인 사용자 입력 요청은 자동으로 변환되지 않으므로 호스트는 자체 연속 정책을 구현해야 합니다.
세션 및 작업 상태 보호
A2AExecutor 는 A2A 컨텍스트 ID를 에이전트 프레임워크 세션 ID로 사용합니다. 어댑터 기반 및 도우미 기반 샘플은 A2A 테넌트와 컨텍스트 ID를 결합하여 애플리케이션에서 선택한 매핑을 보여 줍니다. 모든 접근 방식에서 프로덕션 호스트는 A2A 요청 처리기에 도달하기 전에 호출자를 인증하고, 해당 신뢰할 수 있는 ID에서 테넌트 및 주체를 파생시키고, 모든 작업, 컨텍스트, 연속 및 취소 ID에 권한을 부여해야 합니다.
Important
A2A SDK의 기본 작업 저장소와 푸시 구성 저장소는 인메모리 방식이며, 소유권 범위는 사용자 이름을 기준으로 구분됩니다. 다중 테넌트 서비스의 경우, 동일한 신뢰할 수 있는 테넌트 및 주체로부터 소유권을 가져오는 owner_resolver를 사용하고, 복제본이 다시 시작되거나 스케일 아웃될 수 있는 경우에는 내구성 있는 작업 저장소 및 세션 저장소를 사용합니다.
전체 도우미 기반 서버 및 다중 에이전트 예제는 A2A 호스팅 샘플을 참조하세요. A2A 클라이언트 및 프로토콜 기능은 A2A 에이전트 서비스를 참조하세요.
다음 단계
더 자세히 살펴보기: