Microsoft Bot Framework Streaming (Python)

Warnings

• breaking The Bot Framework Python SDK, including `botframework-streaming`, reached end-of-life with version 4.17.1. It is no longer updated, maintained, or supported through Azure portal service tickets. Customers should consider migrating to alternative solutions like the Microsoft 365 Agents SDK or other AI agent frameworks.
  Fix: Plan for migration to a different bot development framework. Continue using this SDK at your own risk for existing deployments, but be aware of security and compatibility risks.
• gotcha Starting with SDK version 4.15.0, the underlying `aiohttp` package (version 3.9+) requires Python 3.8 or newer. Bots deployed with older Python versions (e.g., 3.7) will need to be updated to Python 3.8+.
  Fix: Ensure your Python environment (local or deployment) is running Python 3.8 or a later compatible version.
• gotcha This `botframework-streaming` library provides low-level streaming protocol implementations. Most bot developers will interact with higher-level abstractions like `botbuilder-core.BotFrameworkAdapter` or `botbuilder-integration-aiohttp.CloudAdapter` for bot development, which leverage this library internally. Direct usage of `botframework-streaming` classes is rare.
  Fix: Focus on using the `botbuilder-core` and `botbuilder-integration-aiohttp` packages for standard bot development. Consult advanced documentation if custom streaming behavior is required.

Install

• pip install botframework-streaming Install latest version

Imports

• StreamingHttpClient

  from botframework.streaming.bot_framework_http_client import StreamingHttpClient

  A low-level client for streaming HTTP requests. Typically not directly used by bot developers.
• StreamingWebSocketClient

  from botframework.streaming.websocket.websocket_client import StreamingWebSocketClient

  A low-level client for WebSocket streaming. Typically not directly used by bot developers.
• StreamingRequestHandler

  from botframework.streaming.streaming_request_handler import StreamingRequestHandler

  Base class for handling incoming streaming requests. Used in implementing custom adapters.

Quickstart

This quickstart demonstrates setting up an `aiohttp` web server to expose a Bot Framework `CloudAdapter`, which internally uses components that leverage `botframework-streaming` for communication. This is the standard way a Python bot would listen for incoming activities, implicitly using streaming capabilities for various channels. Note that the SDK is EOL.

import asyncio
import os
from aiohttp import web
from botbuilder.core import TurnContext, MessageFactory
from botbuilder.integration.aiohttp import CloudAdapter, ConfigurationBotFrameworkAuthentication
from botbuilder.schema import Activity

# NOTE: This example demonstrates how a bot setup would typically use
# components that internally rely on `botframework-streaming` for communication.
# As the SDK is EOL, this is for historical context.

# Create a simple bot for demonstration
class MyEchoBot:
    async def on_turn(self, turn_context: TurnContext):
        if turn_context.activity.type == 'message' and turn_context.activity.text:
            await turn_context.send_activity(MessageFactory.text(f"You said: {turn_context.activity.text}"))
        else:
            await turn_context.send_activity(f"[{turn_context.activity.type} event detected]")

# Configuration and Adapter setup
app_id = os.environ.get('MicrosoftAppId', '') # Get from environment or config
app_password = os.environ.get('MicrosoftAppPassword', '') # Get from environment or config

# Setup Bot Framework Authentication
# For local testing without credentials, pass None for MicrosoftAppId and MicrosoftAppPassword
auth = ConfigurationBotFrameworkAuthentication(
    MicrosoftAppId=app_id,
    MicrosoftAppPassword=app_password
)
adapter = CloudAdapter(auth)

# Instantiate the bot
bot = MyEchoBot()

async def messages(req: web.Request):
    # For authenticated requests (e.g., from Azure Bot Service)
    if 'Authorization' in req.headers:
        response = await adapter.process(req, bot)
        if response:
            return web.Response(status=response.status, headers=response.headers, body=response.body)
    else:
        # Simplified handling for local testing without authentication.
        # In production, authentication is required.
        try:
            activity = await req.json()
            activity = Activity().deserialize(activity)
            turn_context = TurnContext(adapter, activity)
            await bot.on_turn(turn_context)
            return web.Response(status=200)
        except Exception as e:
            print(f"Error processing request: {e}")
            return web.Response(status=500, text=str(e))

    return web.Response(status=202)


async def main():
    web_app = web.Application()
    web_app.router.add_post('/api/messages', messages)
    runner = web.AppRunner(web_app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 3978)
    await site.start()
    print("Bot listening on http://localhost:3978/api/messages")
    while True:
        await asyncio.sleep(3600)

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Bot stopped.")