Proactive message in python

[amitmeel]

In one of our use cases, i want to send the proactive message (notification to the user using some adaptive card or some message from time to time). But this message will be triggered by some external system on /api/notify path. Currently, I don't see any direct way to mount the router to existing Fastpi server (atleast not clear from python documetnation).

currently that's how I have tried to implement the functionality:

import asyncio
import re
import json
from azure.identity import ManagedIdentityCredential
from microsoft_teams.api import MessageActivity, TypingActivityInput
from microsoft_teams.apps import ActivityContext, App
from config import Config
from fastapi import Request, HTTPException
from db import upsert_conversation_id, get_conversation_id

config = Config()

def create_token_factory():
    def get_token(scopes, tenant_id=None):
        credential = ManagedIdentityCredential(client_id=config.APP_ID)
        if isinstance(scopes, str):
            scopes_list = [scopes]
        else:
            scopes_list = scopes
        token = credential.get_token(*scopes_list)
        return token.token
    return get_token

app = App(
    token=create_token_factory() if config.APP_TYPE == "UserAssignedMsi" else None
)
# app = App(tenant_id=config.APP_TENANTID,
#           client_id=config.APP_ID,
#           client_secret=config.APP_PASSWORD
#     )

@app.on_message_pattern(re.compile(r"hello|hi|greetings"))
async def handle_greeting(ctx: ActivityContext[MessageActivity]) -> None:
    """Handle greeting messages."""
    print(ctx.activity)
    user_id = ctx.activity.from_.name
    conversation_id = ctx.activity.conversation.id
    upsert_conversation_id(user_id,conversation_id)
    await ctx.send("Hello! How can I assist you today?")


@app.on_message
async def handle_message(ctx: ActivityContext[MessageActivity]):
    """Handle message activities using the new generated handler system."""
    await ctx.reply(TypingActivityInput())
    user_id = ctx.activity.from_.name
    conversation_id = ctx.activity.conversation.id
    upsert_conversation_id(user_id,conversation_id)

    if "reply" in ctx.activity.text.lower():
        await ctx.reply("Hello! How can I assist you today?")
    else:
        await ctx.send(f"You said '{ctx.activity.text}'")

# Register a custom endpoint on the same FastAPI app
@app.http.post("/api/notify")
async def notify_endpoint(request: Request):
    """
    Simple notification endpoint.
    
    Accepts any JSON or text and sends it to Teams.
    
    Usage:
        # Send text
        curl -X POST http://localhost:3978/api/notify \
             -H "Content-Type: text/plain" \
             -d "Hello from notification!"
        
        # Send JSON with message
        curl -X POST http://localhost:3978/api/notify \
             -H "Content-Type: application/json" \
             -d '{"message": "Test notification"}'
        
        # Send JSON with conversation_id and message
        curl -X POST http://localhost:3978/api/notify \
             -H "Content-Type: application/json" \
             -d '{"conversation_id": "19:xxx", "message": "Targeted notification"}'
    """
    
    # Get content type
    content_type = request.headers.get("content-type", "")
    
    # Parse request body
    if "application/json" in content_type:
        try:
            data = await request.json()
            print(f"📨 Received JSON: {data}")
            
            user_id = "Alex Wilber"
            # Extract message and optional conversation_id
            conversation_id = get_conversation_id(user_id)
            await app.send(conversation_id, str(data))
            
        except Exception as e:
            print(f"❌ Error parsing JSON: {e}")
            raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}")
    
    else:
        # Treat as plain text
        body = await request.body()
        message = body.decode('utf-8')
        print(f"📨 Received text: {message}")
        user_id = "Alex Wilber"
        conversation_id = get_conversation_id(user_id)
        await app.send(conversation_id, message)

if __name__ == "__main__":
    asyncio.run(app.start())

Note: I'm currently running the app using VS Code 365 agent toolkit and 365 agent playground. But when using the external system, I'm hitting this URL ("/api/notify"), and I'm getting an app not started error. When I debug the issue, it is related to the bot credential. Do we have a way using which we don't need to register on Azure Bot Service and get the bot_id, secret etc.

Basically, when I'm using `app.send(conversation_id, messageactivtiy), it checks for app.id (app.credentials basically). Also, how do you secure this URL since this will be a public URL from the teams side once deployed, so that not everyone can hit it? Do we have any way directly in the SDK.

Also, for security purposes, will this SDK automatically take care of authentication, or do we need to manually take care of things since many of the things are not clear from the documentation, and we don't have api reference for this SDK?

[amitmeel]

If anyone from the team or community can provide any suggestion/solution, it would be a big help. Thanks

[Benjiiim]

This PR from @heyitsaamir might improve things in the proactive notification area: microsoft/teams.py#249

Other than better code organization, the main thing this lets us do is not require the app to run to be able to send proactive messages. This is a huge plus point because now the App can be used in scenarios where it doesn't necessarily need to listen to incoming messages (like agentic notifications!)