list_threads

Instructions

Implementation Reference

• src/mcp_teams_server/__init__.py:153-165 (handler)

  MCP tool handler function 'list_threads' - accepts 'limit' (default 50) and 'cursor' (optional) parameters, delegates to TeamsClient.read_threads()

  @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/__init__.py:153-153 (registration)

  Tool registration via @mcp.tool decorator on the FastMCP instance 'mcp' (created at line 88)

  @mcp.tool(name="list_threads", description="List threads in channel with pagination")

• src/mcp_teams_server/teams.py:303-354 (helper)

  TeamsClient.read_threads() - the actual logic using Microsoft Graph API to list channel messages with pagination support (limit, cursor)

  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/teams.py:62-69 (schema)

  PagedTeamsMessages schema - the return type for list_threads, containing cursor, limit, total, and items (list of TeamsMessage)

  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/__init__.py:103-105 (helper)

  Helper function _get_teams_client() used by list_threads to obtain the TeamsClient from the request context

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