botframework-streaming

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