send-notification
Send Markdown-formatted notifications to Microsoft Teams channels to alert users about project updates or important messages.
Instructions
Send a notification message to the user. Supports markdown formatting for messages. Use backticks for code blocks and inline code. Use square brackets for placeholders.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| message | Yes | ||
| project | Yes |
Implementation Reference
- The primary tool handler decorated with @server.call_tool(), which handles calls to 'send-notification'. It validates arguments, checks environment variables, acquires an authentication token, and invokes the send_notification helper to deliver the message.@server.call_tool() async def handle_call_tool( name: str, arguments: dict | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: if name != TOOL_NAME: raise ValueError(f"Unknown tool: {name}") if not arguments: raise ValueError("Missing arguments") message = arguments.get("message") project = arguments.get("project") if not message or not project: raise ValueError("Missing message or project") env_vars, missing_vars = validate_environment_variables() if missing_vars: return [ types.TextContent( type="text", text=f"Missing required environment variables: {', '.join(missing_vars)}" ) ] try: access_token, error = await get_auth_token( env_vars["MICROSOFT_APP_ID"], env_vars["MICROSOFT_APP_PASSWORD"], env_vars["MICROSOFT_APP_TENANT_ID"] ) if error: return [ types.TextContent( type="text", text=f"Authentication failed: {error}" ) ] success, error_msg = await send_notification( env_vars["BOT_ENDPOINT"], access_token, env_vars["EMAIL"], message, project ) if not success: return [ types.TextContent( type="text", text=f"Failed to send notification: {error_msg}" ) ] return [ types.TextContent( type="text", text=f"Sent notification message for project '{project}' with content: {message}", ) ] except Exception as e: return [ types.TextContent( type="text", text=f"Error sending notification: {str(e)}" ) ]
- src/microsoft_teams_mcp/server.py:112-128 (registration)The @server.list_tools() handler that registers the 'send-notification' tool, providing its name (TOOL_NAME), description, and input schema.@server.list_tools() async def handle_list_tools() -> list[types.Tool]: return [ types.Tool( name=TOOL_NAME, description="Send a notification message to the user. Supports markdown formatting for messages. Use backticks for code blocks and inline code. Use square brackets for placeholders.", inputSchema={ "type": "object", "properties": { "message": {"type": "string"}, "project": {"type": "string"}, }, "required": ["message", "project"], }, ) ]
- The JSON schema defining the input parameters for the 'send-notification' tool: message (string, required) and project (string, required).inputSchema={ "type": "object", "properties": { "message": {"type": "string"}, "project": {"type": "string"}, }, "required": ["message", "project"], },
- Core helper function send_notification that sends an HTTP POST request to the configured bot endpoint with the notification payload, using the provided access token.async def send_notification( bot_endpoint: str, access_token: str, email: Optional[str], message: str, project: str ) -> Tuple[bool, Optional[str]]: """ Send notification to the Teams bot endpoint. Args: bot_endpoint: The bot endpoint URL access_token: Authentication token email: User email (optional) message: Notification message project: Project name Returns: Tuple containing (success, error_message) """ try: payload = { "email": email, "message": message, "project": project, } headers = { "Authorization": f"Bearer {access_token}", "Content-Type": "application/json" } async with aiohttp.ClientSession() as session: async with session.post(bot_endpoint, json=payload, headers=headers) as response: if response.status >= 400: response_text = await response.text() return False, f"HTTP {response.status} - {response_text}" return True, None except Exception as e: return False, str(e)
- Helper function get_auth_token that acquires an MSAL access token using client credentials flow for authenticating with Microsoft services.async def get_auth_token(app_id: str, app_password: str, tenant_id: str) -> Tuple[Optional[str], Optional[str]]: """ Get authentication token using MSAL client credentials flow. Args: app_id: The application ID app_password: The application password/secret tenant_id: The tenant ID Returns: Tuple containing (access_token, error_message) """ try: app = msal.ConfidentialClientApplication( client_id=app_id, client_credential=app_password, authority=f"https://login.microsoftonline.com/{tenant_id}" ) scopes = [f"{app_id}/.default"] result = app.acquire_token_for_client(scopes) if "access_token" not in result: error_msg = result.get("error_description", "Failed to acquire token") return None, error_msg return result["access_token"], None except Exception as e: return None, str(e)