#
tokens: 2230/50000 6/6 files
lines: off (toggle) GitHub
raw markdown copy
# Directory Structure

```
├── .gitignore
├── .python-version
├── pyproject.toml
├── README.md
├── src
│   └── microsoft_teams_mcp
│       ├── __init__.py
│       └── server.py
└── uv.lock
```

# Files

--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------

```
3.10

```

--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------

```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

.env

```

--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------

```markdown
# microsoft-teams-mcp MCP server

An MCP Server with a tool for Microsoft Teams chat notifications.

> [!WARNING]  
> This is provided for reference and wasn't tested with MCP clients other than VS Code.

## Components

### Tools

The server implements one tool:
- send-notification: Sends a notification message to Microsoft Teams
  - Takes "message" and "project" as required string arguments
  - Supports Markdown formatting for messages
  - Uses Azure AD authentication to securely communicate with Teams

## Configuration

This requires a Microsoft Teams bot to use for the notifications. You can use [my example Notification Bot](https://github.com/therealjohn/TeamsNotificationBotMCP) created with [Teams Toolkit](https://learn.microsoft.com/en-us/microsoftteams/platform/toolkit/teams-toolkit-fundamentals).

The server requires the following environment variables to be set:

- `BOT_ENDPOINT`: The URL endpoint of your Microsoft Teams bot
- `MICROSOFT_APP_ID`: Application (client) ID from Azure AD app registration
- `MICROSOFT_APP_PASSWORD`: Client secret from Azure AD app registration
- `MICROSOFT_APP_TENANT_ID`: Your Azure AD tenant ID
- `EMAIL`: The email address for the user receiving notifications

You can set these in a `.env` file in the project root directory.

## Quickstart

### Install

#### VS Code

This was tested using MCP support in VS Code, which at the time of creating this was available only in VS Code Insiders.

Add this to the VS Code Insiders Settings (JSON)

```
"mcp": {
  "inputs": [],
  "servers": {
      "MicrosoftTeams": {
          "command": "uv",
          "args": [
              "--directory",
              "<path/to/the/project>/microsoft-teams-mcp",
              "run",
              "microsoft-teams-mcp"
          ],
          "env": {
              "BOT_ENDPOINT": "<endpoint or dev tunnel URL of Teams bot>/api/notification",
              "MICROSOFT_APP_ID": "<microsoft-entra-client-id>",
              "MICROSOFT_APP_PASSWORD": "<microsoft-entra-client-secret>",
              "MICROSOFT_APP_TENANT_ID": "<microsoft-entra-tenant-id>",
              "EMAIL": "<your-email-in-teams>",
          }
      }
  }
    }
```

## Development

### Building

To prepare the package for distribution:

1. Sync dependencies and update lockfile:
```bash
uv sync
```

2. Build package distributions:
```bash
uv build
```

```

--------------------------------------------------------------------------------
/src/microsoft_teams_mcp/__init__.py:
--------------------------------------------------------------------------------

```python
from . import server
import asyncio

def main():
    """Main entry point for the package."""
    asyncio.run(server.run_server())

__all__ = ['main', 'server']
```

--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------

```toml
[project]
name = "microsoft-teams-mcp"
version = "0.1.0"
description = "An MCP Server with tools for Microsoft Teams chat interactions and notifications."
readme = "README.md"
requires-python = ">=3.10"
dependencies = [ "mcp>=1.4.1", "python-dotenv", "aiohttp", "msal"]
[[project.authors]]
name = "John Miller"
email = "[email protected]"

[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"

[project.scripts]
microsoft-teams-mcp = "microsoft_teams_mcp:main"

```

--------------------------------------------------------------------------------
/src/microsoft_teams_mcp/server.py:
--------------------------------------------------------------------------------

```python
import asyncio
import os
import msal

from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
from dotenv import load_dotenv
import aiohttp
from typing import Dict, List, Optional, Tuple

load_dotenv()

SERVER_NAME = "microsoft-teams-mcp"
SERVER_VERSION = "0.1.0"
TOOL_NAME = "send-notification"
REQUIRED_ENV_VARS = ["BOT_ENDPOINT", "MICROSOFT_APP_ID", "MICROSOFT_APP_PASSWORD", "MICROSOFT_APP_TENANT_ID", "EMAIL"]

server = Server(SERVER_NAME)

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)

def validate_environment_variables() -> Tuple[Dict[str, str], List[str]]:
    """
    Validate required environment variables.
    
    Returns:
        Tuple containing (env_vars_dict, missing_vars_list)
    """
    env_vars = {}
        
    for var_name in REQUIRED_ENV_VARS:
        env_vars[var_name] = os.getenv(var_name)
        
    missing_vars = [var_name for var_name in REQUIRED_ENV_VARS if not env_vars[var_name]]
            
    return env_vars, missing_vars

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)

@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"],
            },
        )
    ]

@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)}"
            )
        ]

async def run_server():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name=SERVER_NAME,
                server_version=SERVER_VERSION,
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(run_server())
```