روبوتات Telegram ذاتية الاستضافة

Note

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

Note

لا يتوفر حاليا مساعدو الاستضافة الذاتية لروبوتات Telegram ل Go.

agent-framework-hosting-telegram تحويل تحديثات واجهة برمجة تطبيقات Telegram Bot إلى قيم تشغيل إطار عمل العامل وعرض عمليات التشغيل النهائية أو المتدفقة كعمليات Bot API. لا يوفر عميل روبوت أو وقت تشغيل الاستقصاء أو موجه خطاف الويب أو سجل الأوامر أو إطار عمل التسليم.

pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-telegram azure-identity

استخدم أي مكتبة عميل Telegram يمكنها توفير حمولة تحديث وتنفيذ العمليات التي يرجعها المساعدون. يستخدم aiogramالنموذج ، ولكن المساعدين غير مرتبطين به.

معالجة تحديث

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

async def handle_update(update: Mapping[str, Any]) -> None:
    """Process one Telegram update through the sample agent."""
    callback_query_id = telegram_callback_query_id(update)
    if callback_query_id is not None:
        await bot.answer_callback_query(callback_query_id=callback_query_id)

    chat_id = telegram_chat_id(update)
    session_id = telegram_session_id(update, bot_id=bot.id)
    if chat_id is None or session_id is None:
        return

    # Background webhook tasks may overlap. Serialize each chat so /new cannot
    # delete a session while an earlier response is still updating it.
    async with session_locks.setdefault(session_id, asyncio.Lock()):
        if (command := telegram_command(update)) is not None and await handle_command(update, command):
            return

        async def resolve_file_url(file_id: str) -> str | None:
            file = await bot.get_file(file_id)
            if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES):
                return None
            destination = BytesIO()
            await bot.download_file(file.file_path, destination=destination)
            data = destination.getvalue()
            if len(data) > MAX_MEDIA_BYTES:
                return None
            encoded = base64.b64encode(data).decode("ascii")
            return f"data:application/octet-stream;base64,{encoded}"

        try:
            run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True)
        except ValueError:
            LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True)
            return

        await bot.send_chat_action(chat_id=chat_id, action="typing")
        placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT)

        target = await state.get_target()
        # Reuse one AgentSession per Telegram chat. The /new command removes this
        # mapping so get_or_create_session creates a clean session next time.
        session = await state.get_or_create_session(session_id)
        stream = target.run(
            run["messages"],
            stream=True,
            session=session,
            options=run["options"],
        )
        if not isinstance(stream, ResponseStream):
            raise RuntimeError("agent did not return a response stream")

        last_edit_at = 0.0
        async for operation in telegram_from_streaming_run(
            stream,
            chat_id=chat_id,
            message_id=placeholder.message_id,
            initial_text=PLACEHOLDER_TEXT,
        ):
            if operation["method"] == "editMessageText":
                delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at)
                if delay > 0:
                    await asyncio.sleep(delay)
                last_edit_at = time.monotonic()
            await execute_operation(operation)

        # Persist the updated AgentSession back under the stable per-chat key after
        # streaming has finalized and the history provider has recorded the turn.
        await state.set_session(session_id, session)

للاستطلاع وإعداد خطاف الويب ومعالجة الأوامر ونهج الوسائط الواردة وتحرير الدفق وإرشادات نشر الإنتاج، راجع عينة Telegram المحلية.

Important

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

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

إضافة A2A

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

• نظرة عامة على الاستضافة الذاتية
• استجابات OpenAI
• MCP
• وكلاء مستضافون في Foundry