A2A-агенты с самостоятельным размещением

Используйте .NET пакеты размещения A2A для предоставления агента Agent Framework через ASP.NET Core. Полный справочник по серверу с несколькими языками см. в разделе агентов узлов с помощью A2A .

Используйте пакет Go provider/a2aprovider с официальными обработчиками сервера A2A для Go. Полный пример сервера см. в разделе "Агенты узла" с помощью A2A .

Agent Framework предоставляет два пакета Python для размещения агентов и рабочих процессов через официальный пакет SDK A2A:

Package	Модель интеграции	Используйте его, когда
`agent-framework-a2a`	Компонент с жёстко заданной архитектурой `A2AExecutor`, который преобразует запросы, запускает агента и публикует события и артефакты задач A2A.	Вам требуется стандартное поведение взаимодействия Agent Framework с A2A, и нужно лишь собрать сервер A2A SDK.
`agent-framework-hosting-a2a`	Инкрементальные компоненты для исполнителя, принадлежащего приложению. Начните с базового конвертера агента или конвертера рабочего процесса, а при необходимости используйте `AgentA2AAdapter` или `WorkflowA2AAdapter`, которые основаны на этих конвертерах и добавляют генерацию нативных карточек и проверку режима.	Приложение должно обеспечивать сопоставление сессий, переходы между задачами, доставку событий, определение границ артефактов, преобразование вывода или работу в качестве многопротокольного хоста.

Оба пакета используют собственные типы пакетов SDK A2A и серверные компоненты. Приложение предоставляет обработчик запросов, хранилище задач, маршруты или построитель приложений SDK, проверку подлинности и развертывание. С помощью agent-framework-hosting-a2aприложения можно создать карточку агента напрямую или разрешить адаптеру создать его.

Использование исполнителя A2A с мнением

Установите agent-framework-a2a, если встроенный адаптер сервера подходит для вашего жизненного цикла:

pip install --pre agent-framework-a2a starlette uvicorn

A2AExecutor реализует пакет AgentExecutor из SDK A2A. Он считывает пользовательский ввод из контекста запроса A2A, создает сеанс Agent Framework на основе идентификатора контекста A2A, запускает агент в потоковом или непотоковом режиме, преобразует поддерживаемые выходные данные и публикует события состояния задачи и артефактов через TaskUpdater SDK.

Соберите его с помощью SDK A2A DefaultRequestHandler, хранилища задач, карточки агента и приложения Starlette или другой поддерживаемой серверной интеграции. Настройте потоковую передачу с помощью A2AExecutor(agent, stream=True), передайте параметры запуска стабильного агента через run_kwargs или создайте подкласс A2AExecutor и переопределите handle_events, если требуется другое сопоставление выходных данных.

A2AExecutor ограничен эндпоинтом A2A и напрямую управляет своим A2A-выполнением и сопоставлением сеансов. Используйте пакеты размещения, если один и тот же агент должен быть доступен через несколько протоколов в одном приложении.

Полную настройку сервера см. в разделе Публикация агента Agent Framework через A2A.

Используйте адаптер в исполнителе, принадлежащем приложению

Установите пакет хостинга, если приложение использует собственный исполнитель A2A, но вы хотите, чтобы Agent Framework создавал публичную карточку и проверял конвертации:

pip install --pre agent-framework-hosting-a2a starlette uvicorn

AgentA2AAdapter поддерживает агент или AgentState. Его асинхронный метод get_card определяет публичное имя и описание, по умолчанию использует консервативные текстовые режимы и может определять нативные навыки A2A из экземпляров Agent Framework SkillsProvider. Возможности сервера и поддерживаемые интерфейсы остаются явными, так как они описывают конечную точку приложения, а не метод агента 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объект и создает исполнителя, принадлежащий приложению, с обработчиком запроса пакета SDK A2A:

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 преобразует A2A Message в аргументы запуска Agent Framework.
a2a_from_run преобразует ответы Agent Framework и обновления в потоковом режиме в значения A2A Part.

Исполнитель выбирает ключи сеанса и владеет переходами задач, очередями событий, идентификаторами артефактов, границами сообщений и исходящей доставкой. a2a_from_run возвращает плоский список частей, чтобы приложение могло сгруппировать эти части в сообщения A2A или артефакты и применять метаданные на уровне сообщения.

Конфигурация хостинга также поддерживает многопротокольные приложения. Использовать один и тот же целевой агент и одну и ту же инфраструктуру AgentState в маршрутах A2A, OpenAI Responses, Telegram и MCP, при этом каждая конечная точка сохраняет собственную политику преобразования, авторизации и ключей сеанса. Это позволяет клиентам обращаться к одному агенту по разным протоколам одновременно, не создавая отдельное развертывание агента для каждой конечной точки.

Скомпонуйте вспомогательные компоненты в нативном исполнителе A2A SDK. Этот пример создает и обновляет задачи A2A, преобразует входящее сообщение в запуск Agent Framework, сохраняет обновленный 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, но вспомогательные функции не привязаны ни к одному из них. Используйте платформу приложений или построитель приложений пакета SDK A2A для обслуживания карточки агента 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 типа text, raw или data для единственного входного типа start-executor в рабочем процессе, а преобразование выходных данных сопоставляет выходные данные завершённого публичного рабочего процесса с собственными частями A2A. Вызывайте get_card перед преобразованием проверенных выходных данных, когда адаптеру необходимо определить режимы вывода.

Приложение по-прежнему отвечает за собственный нативный исполнитель A2A, а также за потоковую передачу сведений о ходе выполнения, статусе задачи, артефактах, контрольных точках и продолжение с участием человека. Ожидающие запросы, требующие участия пользователя, не преобразуются автоматически, поэтому хост должен реализовать собственную политику продолжения.

Безопасные сеансы и состояние задачи

A2AExecutor использует идентификатор контекста A2A в качестве идентификатора сеанса Agent Framework. Примеры на основе адаптера и вспомогательных компонентов объединяют идентификатор арендатора A2A и идентификатор контекста, чтобы продемонстрировать сопоставление, выбранное приложением. В каждом случае промышленный хост должен аутентифицировать вызывающую сторону до того, как запрос попадёт в обработчик запросов A2A, определить тенанта и субъекта на основе этого доверенного удостоверения, а также авторизовать все идентификаторы задач, контекста, продолжения и отмены.

Important

Хранилище задач по умолчанию и хранилища push-конфигураций в SDK A2A находятся в памяти и используют имя пользователя для определения области владения. Для службы с несколькими клиентами используйте объект, производный от того же доверенного клиента и субъекта, и используйте owner_resolver устойчивые хранилища задач и сеансов, когда реплики могут перезапускать или масштабироваться.

Полноценные примеры сервера на основе вспомогательных компонентов и примеры с несколькими агентами см. в примерах хостинга A2A. Сведения о клиентах И протоколах A2A см. в службе агента A2A.

Дальнейшие действия

Агенты узлов с A2A

Вернитесь глубже:

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-08-13