Microsoft Teams MCP Server

Instructions

Implementation Reference

• src/microsoft_teams_mcp/server.py:129-199 (handler)

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

• src/microsoft_teams_mcp/server.py:118-125 (schema)

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

• src/microsoft_teams_mcp/server.py:69-111 (helper)

  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)

• src/microsoft_teams_mcp/server.py:22-52 (helper)

  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)