MCP Microsoft Teams Server

Instructions

Implementation Reference

• src/mcp_teams_server/__init__.py:154-167 (handler)

  MCP tool handler for 'list_threads': registers the tool, validates inputs via Pydantic Fields, retrieves TeamsClient from context, and calls read_threads with pagination parameters.

  @mcp.tool(name="list_threads", description="List threads in channel with pagination")
  async def list_threads(
      ctx: Context,
      limit: int = Field(
          description="Maximum number of items to retrieve or page size", default=50
      ),
      cursor: str | None = Field(
          description="Pagination cursor for the next page of results", default=None
      ),
  ) -> PagedTeamsMessages:
      await ctx.debug(f"list_threads with cursor={cursor} and limit={limit}")
      client = _get_teams_client(ctx)
      return await client.read_threads(limit, cursor)

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

  Pydantic schema/model for the paged response returned by list_threads tool, including cursor, limit, total, and list of TeamsMessage items.

  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:288-339 (helper)

  TeamsClient.read_threads method: the core implementation that fetches channel messages (threads) from Microsoft Graph API using pagination (limit/cursor), maps to PagedTeamsMessages.

  async def read_threads(
      self, limit: int = 50, cursor: str | None = None
  ) -> PagedTeamsMessages:
      """Read all threads in configured teams channel.
  
      Args:
          cursor: The pagination cursor.
  
          limit: The pagination page size
  
      Returns:
          Paged team channel messages containing
      """
      try:
          query = MessagesRequestBuilder.MessagesRequestBuilderGetQueryParameters(
              top=limit
          )
          request = RequestConfiguration(query_parameters=query)
          if cursor is not None:
              response = (
                  await self.graph_client.teams.by_team_id(self.team_id)
                  .channels.by_channel_id(self.teams_channel_id)
                  .messages.with_url(cursor)
                  .get(request_configuration=request)
              )
          else:
              response = (
                  await self.graph_client.teams.by_team_id(self.team_id)
                  .channels.by_channel_id(self.teams_channel_id)
                  .messages.get(request_configuration=request)
              )
  
          result = PagedTeamsMessages(
              cursor=response.odata_next_link,  # pyright: ignore
              limit=limit,
              total=response.odata_count,  # pyright: ignore
              items=[],
          )
          if response.value is not None:  # pyright: ignore
              for message in response.value:  # pyright: ignore
                  result.items.append(
                      TeamsMessage(
                          message_id=message.id,  # pyright: ignore
                          content=message.body.content,  # pyright: ignore
                          thread_id=message.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-105 (helper)

  Helper function to retrieve the TeamsClient instance from the MCP request context, used by list_threads handler.

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

• src/mcp_teams_server/teams.py:44-50 (schema)

  Pydantic model for individual TeamsMessage items in the paged response of list_threads.

  class TeamsMessage(BaseModel):
      thread_id: str = Field(
          description="Thread ID as a string in the format '1743086901347'"
      )
      message_id: str = Field(description="Message ID")
      content: str = Field(description="Message content")