Agen A2A yang dihost sendiri

Gunakan paket hosting .NET A2A untuk mengekspos agen Agent Framework melalui ASP.NET Core. Lihat Integrasi A2A untuk penyiapan paket dan contoh server lengkap.

Gunakan paket Go provider/a2aprovider dengan handler server A2A Go resmi. Lihat Integrasi A2A untuk contoh server lengkap.

Agent Framework menyediakan dua paket Python untuk agen hosting dan alur kerja melalui SDK A2A resmi:

Package Model integrasi Gunakan saat
agent-framework-a2a A2AExecutor dengan pendekatan yang sudah ditetapkan yang mengonversi permintaan, menjalankan agen, dan memublikasikan peristiwa tugas serta artefak A2A. Anda menginginkan perilaku Agent Framework-to-A2A standar dan hanya perlu merakit server A2A SDK.
agent-framework-hosting-a2a Blok penyusun inkremental untuk eksekutor yang dimiliki aplikasi. Mulailah dengan agen dasar atau konverter alur kerja, dan secara opsional gunakan AgentA2AAdapter atau WorkflowA2AAdapter, yang dibangun di atas konverter tersebut untuk menambahkan pembuatan kartu native dan validasi mode. Aplikasi Anda perlu memiliki pemetaan sesi, transisi tugas, pengiriman peristiwa, batas artefak, konversi output, atau host multi-protokol.

Kedua paket menggunakan jenis A2A SDK asli dan komponen server. Aplikasi Anda menyediakan penangan permintaan, penyimpanan tugas, rute atau penyusun aplikasi SDK, autentikasi, dan penyebaran. Dengan agent-framework-hosting-a2a, aplikasi dapat membuat kartu agen secara langsung atau membiarkan adaptor menghasilkannya.

Gunakan eksekutor A2A dengan opini bawaan

Instal agent-framework-a2a saat adaptor server bawaan cocok dengan siklus hidup Anda:

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

A2AExecutor mengimplementasikan A2A SDK AgentExecutor. Membaca input pengguna dari konteks permintaan A2A, membuat sesi Agent Framework dari ID konteks A2A, menjalankan agen dalam mode streaming maupun non-streaming, mengonversi konten keluaran yang didukung, dan memublikasikan status tugas serta event artefak melalui TaskUpdater milik SDK.

Buat dengan aplikasi A2A SDK DefaultRequestHandler, penyimpanan tugas, kartu agen, dan Starlette atau integrasi server lain yang didukung. Konfigurasikan streaming dengan A2AExecutor(agent, stream=True), berikan opsi eksekusi agen yang stabil melalui run_kwargs, atau subkelas A2AExecutor dan ambil alih handle_events saat Anda memerlukan pemetaan output yang berbeda.

A2AExecutor berada dalam cakupan endpoint A2A dan mengelola eksekusi A2A serta pemetaan sesinya secara langsung. Gunakan paket hosting ketika agen yang sama harus tersedia melalui beberapa protokol dalam satu aplikasi.

Untuk konfigurasi server lengkap, lihat Mengekspos agen Agent Framework melalui A2A.

Menggunakan adaptor di eksekutor milik aplikasi

Instal paket hosting saat aplikasi Anda memiliki eksekutor A2A asli tetapi ingin Agent Framework menghasilkan kartu publik dan memvalidasi konversi:

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

AgentA2AAdapter menerima agen atau AgentState. Metode asinkronnya get_card menurunkan nama dan deskripsi publik, menggunakan mode teks konservatif secara bawaan, dan dapat menyimpulkan keterampilan A2A bawaan dari instans Agent Framework SkillsProvider. Kemampuan server dan antarmuka yang didukung tetap eksplisit karena menjelaskan titik akhir aplikasi daripada metode agen run .

Adaptor mengekspos a2a_to_run dan a2a_from_run metode yang memvalidasi nilai terhadap mode kartu yang dikonfigurasi secara default. Aplikasi ini masih memiliki eksekutor A2A, siklus hidup tugas, antrean peristiwa, batas artefak, kebijakan sesi, autentikasi, rute, dan penyebaran.

Pelaksana ini menggunakan satu adaptor untuk konversi masuk, status agen, dan konversi keluar:

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.")]),
            )

Penyiapan server membuat adaptor, menghasilkan AgentCard native-nya, dan menggabungkan eksekutor milik aplikasi dengan penangan permintaan 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,
    )

Membangun eksekutor A2A milik aplikasi

Gunakan pembantu hosting mandiri ketika aplikasi Anda juga memerlukan kontrol langsung atas pembuatan kartu:

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

Pembantunya netral kerangka kerja:

  • a2a_to_run mengonversi argumen eksekusi A2A Message ke Agent Framework.
  • a2a_from_run mengonversi respons Agent Framework dan pembaruan streaming ke nilai A2A Part .

Eksekutor Anda memilih kunci sesi dan bertanggung jawab atas transisi tugas, antrean peristiwa, ID artefak, batas-batas pesan, dan pengiriman keluar. a2a_from_run mengembalikan daftar bagian datar sehingga aplikasi dapat mengelompokkan bagian tersebut ke dalam pesan atau artefak A2A dan menerapkan metadata tingkat pesan.

Pengaturan hosting juga mendukung aplikasi multi-protokol. Gunakan target agen dan infrastruktur AgentState yang sama di semua rute A2A, OpenAI Responses, Telegram, dan MCP, sementara setiap endpoint protokol tetap memiliki kebijakan konversi, otorisasi, dan kunci sesi masing-masing. Ini memungkinkan klien menjangkau satu agen melalui protokol yang berbeda secara bersamaan tanpa membuat penyebaran agen terpisah untuk setiap titik akhir.

Buat pembantu dalam pelaksana A2A SDK asli. Contoh ini membuat dan memperbarui tugas A2A, mengubah pesan masuk menjadi proses Agent Framework, menyimpan sesi AgentState yang diperbarui setelah stream selesai, dan menerbitkan komponen yang dikembalikan sebagai artefak.

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.")]),
            )

Sampel menggunakan Starlette dan Uvicorn, tetapi pembantu juga tidak terikat. Gunakan kerangka kerja aplikasi Anda atau penyusun aplikasi A2A SDK untuk melayani kartu agen A2A dan rute 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, "/"),
    ]
)

Menghosting alur kerja dengan adaptor

WorkflowA2AAdapter menyediakan pembuatan kartu dan batas konversi yang sama untuk alur kerja atau WorkflowState. Ini menyimpulkan mode input dan output konservatif dari jenis alur kerja yang dideklarasikan, atau Anda dapat menyediakan mode eksplisit untuk representasi khusus aplikasi.

Helper mandiri a2a_to_workflow_run dan a2a_from_workflow_run menyediakan konversi input dan output alur kerja bertipe. Adaptor menyediakannya sebagai metode asinkron a2a_to_run dan sinkron a2a_from_run yang memvalidasi berdasarkan mode kartu yang berlaku. Konversi input menerima satu bagian A2A berjenis teks, mentah, atau data untuk jenis input start-executor tunggal milik alur kerja, dan konversi output memetakan output publik alur kerja yang telah selesai ke bagian A2A native. Panggil get_card sebelum konversi output yang divalidasi ketika adaptor harus menyimpulkan mode output.

Aplikasi ini tetap bertanggung jawab atas pelaksana A2A asli dan untuk kemajuan streaming, status tugas, artefak, titik pemeriksaan, dan kelanjutan human-in-the-loop. Permintaan input manusia yang tertunda tidak dikonversi secara otomatis, sehingga host harus menerapkan kebijakan kelanjutannya sendiri.

Sesi aman dan status tugas

A2AExecutor menggunakan ID konteks A2A sebagai ID sesi Agent Framework. Sampel berbasis adapter dan helper menggabungkan tenant A2A dan ID konteks untuk menunjukkan pemetaan yang dipilih oleh aplikasi. Dalam setiap pendekatan, host produksi harus mengautentikasi pemanggil sebelum permintaan mencapai penangan permintaan A2A, memperoleh tenant dan subjek dari identitas tepercaya tersebut, serta mengautorisasi semua ID tugas, konteks, kelanjutan, dan pembatalan.

Important

Penyimpanan tugas default dan konfigurasi push SDK A2A disimpan dalam memori, dan kepemilikannya dicakup berdasarkan nama pengguna. Untuk layanan multitenan, gunakan owner_resolver yang mewarisi kepemilikan dari penyewa dan subjek tepercaya yang sama, dan gunakan penyimpanan tugas dan sesi yang persisten saat replika dapat dimulai ulang atau ditingkatkan skalanya.

Untuk contoh server berbasis pembantu dan multi-agen lengkap, lihat sampel hosting A2A. Untuk klien A2A dan kemampuan protokol, lihat Integrasi A2A.

Langkah berikutnya

Pelajari tentang integrasi A2A

Masuk lebih dalam:

  • Last updated on