عوامل المضيف الذاتي كأدوات MCP

Note

سيتوفر دعم أداة MCP ذاتية الاستضافة في .NET قريبا.

Note

دعم أداة MCP ذاتية الاستضافة غير متوفر حاليا ل Go.

يستخدم agent-framework-hosting-mcp لعرض عامل إطار عمل العامل أو سير العمل كأداة على SDK بروتوكول سياق النموذج الأصلي. لا تختار الحزمة إطار عمل ويب أو التفاف دورة حياة خادم MCP SDK؛ لا يزال التطبيق الخاص بك يمتلك Serverتسجيل المعالج والنقل ونهج مفتاح الجلسة والمصادقة والتخويل والنشر.

pip install --pre agent-framework-hosting-mcp

التحويل عند حد البروتوكول

mcp_to_run(...) يحول وسيطات أداة MCP التي تم التحقق من صحتها إلى رسائل إطار عمل العامل وخيارات الدردشة المحددة، ويحول mcp_from_run(...) استجابة مكتملة إلى قيم MCP ContentBlock أصلية. استخدم هاتين الدالتين مباشرة عندما يحتاج عقد أداة التطبيق إلى مخطط ومعالج أصلي مخصص بالكامل:

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Return the app-owned native MCP tool definition."""
    return [
        types.Tool(
            name="run_agent_manually",
            description=agent.description or "",
            inputSchema={
                "type": "object",
                "properties": {
                    TASK_ARGUMENT: {
                        "type": "string",
                        "description": "The request for the hosted agent.",
                    },
                    **CHAT_OPTION_ARGUMENTS,
                },
                "required": [TASK_ARGUMENT],
                "additionalProperties": False,
            },
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Convert, run, and render without the agent-backed adapter."""
    if name != "run_agent_manually":
        raise ValueError(f"Unknown MCP tool: {name}")
    run = mcp_to_run(
        arguments,
        argument_name=TASK_ARGUMENT,
        chat_option_arguments=CHAT_OPTION_ARGUMENTS,
    )
    result = await agent.run(run["messages"], options=run["options"])
    return mcp_from_run(result)

يتم نسخ أسماء الوسيطات المدرجة في chat_option_arguments فقط في run["options"]؛ تظل وسيطات MCP الأخرى متوفرة على التمثيل الأولي للرسالة ولكن لا تتم إعادة توجيهها إلى عميل النموذج.

استضافة عامل كأداة واحدة تم إنشاؤها

AgentMCPTool اشتقاق اسم الأداة الأصلية والوصف والمخطط من عامل، ويحافظ على محاذاة سرد وتحليل وتنفيذ وتحويل النتائج بحيث لا يمكن للاثنين الانجراف:

agent_tool = AgentMCPTool(
    agent,
    name="run_agent",
    argument_description="The request for the hosted agent.",
    chat_option_parameters={
        "reasoning_effort": {
            "type": "string",
            "enum": ["low", "medium", "high"],
            "description": "Optional reasoning effort for models that support it.",
        }
    },
)


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Describe the app-owned MCP tool schema."""
    return await agent_tool.list_tools()


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Run the app-owned tool with native MCP and Agent Framework values."""
    return await agent_tool.call_tool(name, arguments)

AgentMCPTool يستخدم اسم العامل ووصفه ما لم يتم تجاوزه. parameters يضيف خصائص مخطط JSON المملوكة للتطبيق والتي تظل متوفرة في وسيطات MCP الأولية، ويضيف chat_option_parameters خصائص يتم نسخ قيمها بشكل صريح إلى خيارات دردشة إطار عمل العامل.

استمرار جلسة عمل لكل مكالمة

قم بتمرير قائمة AgentState و session_id_parameter للسماح للمكالمات المتكررة بنفس المحادثة المعتمة والمحددة session_id من قبل التطبيق بمتابعة محادثة واحدة:

session_locks: dict[str, asyncio.Lock] = {}


@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Return the agent-derived MCP tool definition."""
    return await agent_tool.list_tools()


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, object] | None) -> list[types.ContentBlock]:
    """Serialize calls per app-owned session before using ``AgentState``."""
    session_id = arguments.get("session_id") if arguments else None
    if not isinstance(session_id, str) or not session_id:
        raise ValueError("MCP tool argument 'session_id' must be a non-empty string.")
    lock = session_locks.setdefault(session_id, asyncio.Lock())
    async with lock:
        return await agent_tool.call_tool(name, arguments)

AgentMCPTool ينفذ AgentState فقط تسلسل الحصول على/تشغيل/تعيين جلسة العمل؛ يجب أن يقوم التطبيق الخاص بك بمصادقة أو تخويل معرف الجلسة وتسلسل المكالمات المتزامنة لنفس الجلسة، كما يفعل النموذج مع كل جلسة asyncio.Lockعمل. هذا ليس previous_response_idتفريعا بنمط - يجب أن يقبل التطبيق الذي يحتاج إلى تفرع محادثة معرفات المصدر والوجهة المنفصلة، ونسخ جلسة المصدر، وتخزين النتيجة ضمن مفتاح الوجهة.

استضافة سير عمل كأداة

WorkflowMCPTool اشتقاق أداة MCP أصلية واحدة من نوع إدخال منفذ سير العمل وتحويل مخرجات سير العمل المكتملة. تصبح Dataclass وPydantic وغيرها من المدخلات على شكل كائن وسيطات MCP من المستوى الأعلى؛ يتم تضمين المدخلات الأولية في اسم وسيطة قابل للتكوين:

server = Server("agent-framework-hosting-mcp-workflow-sample")
workflow_tool = WorkflowMCPTool(
    WorkflowState(create_workflow, cache_target=False),
    name="draft_content",
)

تحافظ مثيلات سير العمل على حالة التنفيذ، لذلك يجب أن توفر التطبيقات التي تحتاج إلى استدعاءات مستقلة مصنعا WorkflowState ب cache_target=False، كما هو موضح أعلاه. تظل استعادة نقاط التحقق والاستجابات البشرية في الحلقة ومعرفات المتابعة مملوكة للتطبيق؛ إذا طلب سير العمل إدخالا خارجيا، يرفع المحول بدلا من إرجاع نتيجة أداة ناجحة فارغة.

للحصول على المجموعة الكاملة من الخوادم القابلة للتشغيل — بما في ذلك متغير FastMCP الذي يستمد مخططه من دالة مزينة — راجع عينات استضافة MCP.

Important

تعامل مع معرف جلسة MCP وأي وسيطة معرفة session_id من قبل التطبيق كإدخل غير موثوق به. قم بمصادقة المتصل وتخويله قبل استخدام إما لتحميل حالة جلسة العمل أو حفظها، واشتقاق التقسيم الدائم من المستأجر أو المستخدم أو مساحة العمل المصادق عليها بدلا من القيمة الأولية.

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

تعرف على استضافة A2A

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

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