عوامل A2A ذاتية الاستضافة

استخدم حزم استضافة .NET A2A لعرض عامل إطار عمل العامل من خلال ASP.NET Core. راجع عوامل المضيف مع A2A للحصول على دليل خادم كامل متعدد اللغات.

استخدم حزمة Go provider/a2aprovider مع معالجات خادم A2A Go الرسمية. راجع عوامل المضيف مع A2A للحصول على مثال خادم كامل.

يوفر إطار عمل العامل حزمتين Python لوكلاء الاستضافة ومهام سير العمل من خلال A2A SDK الرسمي:

الحزمة	نموذج التكامل	استخدمه عندما
agent-framework-a2a	تم الفتوى A2AExecutor التي تحول الطلبات، وتشغل عامل، وتنشر أحداث مهمة A2A والبيانات الاصطناعية.	تريد سلوك إطار عمل العامل القياسي إلى 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، وينشئ جلسة عمل Agent Framework من معرف سياق A2A، ويشغل العامل في وضع الدفق أو غير المتدفق، ويحول محتوى الإخراج المدعوم، وينشر حالة المهمة وأحداث البيانات الاصطناعية من خلال 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 الاسم العام والوصف، ويستخدم أوضاع نصية متحفظة بشكل افتراضي، ويمكن أن يستنتج مهارات A2A الأصلية من مثيلات إطار عمل SkillsProvider العامل. تظل قدرات الخادم والواجهات المدعومة صريحة لأنها تصف نقطة نهاية التطبيق بدلا من أسلوب العامل run .

يعرض a2a_to_runa2a_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 تحويل A2A Message إلى وسيطات تشغيل إطار عمل العامل.
• a2a_from_run تحويل استجابات إطار عمل العامل وتحديثات الدفق إلى قيم A2A Part .

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

يدعم إعداد الاستضافة أيضا التطبيقات متعددة البروتوكولات. شارك نفس هدف العامل والبنية AgentState الأساسية عبر مسارات A2A واستجابات OpenAI و Telegram و MCP، بينما تحتفظ كل نقطة نهاية بروتوكول بنهج التحويل والتخويل ومفتاح الجلسة الخاص بها. يتيح هذا للعملاء الوصول إلى عامل واحد من خلال بروتوكولات مختلفة في نفس الوقت دون إنشاء نشر وكيل منفصل لكل نقطة نهاية.

إنشاء المساعدين في منفذ 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 وOvicorn، ولكن المساعدين غير مرتبطين بأي منهما. استخدم إطار عمل التطبيق الخاص بك أو منشئ تطبيق 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 الأصلي وعن تقدم الدفق وحالة المهمة والبيانات الاصطناعية ونقاط التحقق والمتابعة البشرية في الحلقة. لا يتم تحويل طلبات الإدخال البشري المعلقة تلقائيا، لذلك يجب على المضيف تنفيذ نهج المتابعة الخاص به.

تأمين جلسات العمل وحالة المهمة

A2AExecutor يستخدم معرف سياق A2A كمعرف جلسة عمل Agent Framework. تجمع العينات المستندة إلى المحول والمستندة إلى المساعد بين مستأجر A2A ومعرف السياق لإظهار تعيين محدد من قبل التطبيق. في كل نهج، يجب على مضيف الإنتاج مصادقة المتصل قبل أن يصل إلى معالج طلب A2A، واشتقاق المستأجر والموضوع من تلك الهوية الموثوق بها، وتخويل جميع معرفات المهمة والسياق والمتابعة والإلغاء.

Important

المهمة الافتراضية ل A2A SDK ومخازن تكوين الدفع في الذاكرة وملكية النطاق حسب اسم المستخدم. بالنسبة لخدمة متعددة المستأجرين، استخدم owner_resolver الذي يستمد الملكية من نفس المستأجر والموضوع الموثوق به، واستخدم مخازن المهام والجلسات الدائمة عندما يمكن إعادة تشغيل النسخ المتماثلة أو توسيع نطاقها.

للحصول على خادم كامل قائم على المساعد وأمثلة متعددة العوامل، راجع عينات استضافة A2A. للحصول على عملاء A2A وقدرات البروتوكول، راجع خدمة عامل A2A.

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

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

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

• نظرة عامة على الاستضافة الذاتية
• استجابات OpenAI
• برقيه