MCP Microsoft Teams Server

Instructions

Implementation Reference

• src/mcp_teams_server/__init__.py:142-152 (handler)

  MCP tool handler function for 'read_thread' that logs the request, retrieves the TeamsClient instance, and delegates to client.read_thread_replies to fetch thread replies.

  @mcp.tool(name="read_thread", description="Read replies in a thread")
  async def read_thread(
      ctx: Context,
      thread_id: str = Field(
          description="The thread ID as a string in the format '1743086901347'"
      ),
  ) -> PagedTeamsMessages:
      await ctx.debug(f"read_thread with thread_id={thread_id}")
      client = _get_teams_client(ctx)
      return await client.read_thread_replies(thread_id, 50)

• src/mcp_teams_server/teams.py:59-66 (schema)

  Pydantic model defining the output schema for paginated Teams messages, used as return type for read_thread tool.

  class PagedTeamsMessages(BaseModel):
      cursor: str | None = Field(
          description="Cursor to retrieve the next page of messages."
      )
      limit: int = Field(description="Page limit, maximum number of items to retrieve")
      total: int = Field(description="Total items available for retrieval")
      items: list[TeamsMessage] = Field(description="List of channel messages or threads")

• src/mcp_teams_server/teams.py:341-396 (helper)

  Core helper method in TeamsClient that performs the Microsoft Graph API query to retrieve paginated replies for a specific thread ID, mapping responses to PagedTeamsMessages.

  async def read_thread_replies(
      self, thread_id: str, limit: int = 50, cursor: str | None = None
  ) -> PagedTeamsMessages:
      """Read all replies in a thread.
  
      Args:
          thread_id: Thread ID to read
          cursor: The pagination cursor
          limit: The pagination page size
  
      Returns:
          List of thread messages
      """
      try:
          params = RepliesRequestBuilder.RepliesRequestBuilderGetQueryParameters(
              top=limit
          )
          request = RequestConfiguration(query_parameters=params)
  
          if cursor is not None:
              replies = (
                  await self.graph_client.teams.by_team_id(self.team_id)
                  .channels.by_channel_id(self.teams_channel_id)
                  .messages.by_chat_message_id(thread_id)
                  .replies.with_url(cursor)
                  .get(request_configuration=request)
              )
          else:
              replies = (
                  await self.graph_client.teams.by_team_id(self.team_id)
                  .channels.by_channel_id(self.teams_channel_id)
                  .messages.by_chat_message_id(thread_id)
                  .replies.get(request_configuration=request)
              )
  
          result = PagedTeamsMessages(
              cursor=cursor,
              limit=limit,
              total=replies.odata_count,  # pyright: ignore
              items=[],
          )
  
          if replies is not None and replies.value is not None:
              for reply in replies.value:
                  result.items.append(
                      TeamsMessage(
                          message_id=reply.id,  # pyright: ignore
                          content=reply.body.content,  # pyright: ignore
                          thread_id=reply.reply_to_id,  # pyright: ignore
                      )
                  )
  
          return result
      except Exception as e:
          LOGGER.error(f"Error reading thread: {str(e)}")
          raise

• src/mcp_teams_server/__init__.py:104-106 (helper)

  Utility function to retrieve the TeamsClient instance from the MCP context, used by the read_thread handler.

  def _get_teams_client(ctx: Context) -> TeamsClient:
      return ctx.request_context.lifespan_context.client