Twist MCP Server

Instructions

Implementation Reference

• src/threads.py:34-83 (handler)

  Handler function that implements the core logic of twist_threads_get tool by calling Twist API endpoint threads/get with provided parameters.

  def twist_threads_get(
      ctx: Context,
      channel_id: int,
      as_ids: Optional[bool] = None,
      filter_by: Optional[str] = None,
      limit: Optional[int] = None,
      newer_than_ts: Optional[int] = None,
      older_than_ts: Optional[int] = None,
      before_id: Optional[int] = None,
      after_id: Optional[int] = None,
      workspace_id: Optional[int] = None,
      is_pinned: Optional[bool] = None,
      is_starred: Optional[bool] = None,
      order_by: Optional[str] = None,
      exclude_thread_ids: Optional[List[int]] = None
  ) -> Union[str, List[Dict[str, Any]]]:
      """Gets all threads in a channel.
  
      Args:
          channel_id: The id of the channel
          as_ids: If enabled, only the ids of the threads are returned
          filter_by: A filter can be one of "attached_to_me" or "everyone". Default is "everyone"
          limit: Limits the number of threads returned (default is 20, maximum is 500)
          newer_than_ts: Limits threads to those newer when the specified Unix time
          older_than_ts: Limits threads to those older when the specified Unix time
          before_id: Limits threads to those with a lower than the specified id
          after_id: Limits threads to those with a higher than the specified id
          workspace_id: The id of the workspace
          is_pinned: If enabled, only pinned threads are returned
          is_starred: If enabled, only starred threads are returned
          order_by: The order of the threads returned. Either "desc" (default) or "asc"
          exclude_thread_ids: The thread ids that should be excluded from the results
      """
      all_params = locals()
      token = ctx.request_context.lifespan_context.twist_token
      params = {k: v for k, v in all_params.items() if k != 'ctx' and v is not None}
  
      try:
          logger.info(f"Getting threads for channel ID: {channel_id}")
          threads_data = twist_request("threads/get", params=params, token=token)
  
          if not threads_data:
              logger.info("No threads found")
              return "No threads found"
  
          logger.info(f"Retrieved {len(threads_data)} threads")
          return threads_data
      except Exception as error:
          logger.error(f"Error getting threads: {error}")
          return f"Error getting threads: {str(error)}"

• main.py:42-48 (registration)

  Dynamic registration code that registers all twist_ prefixed functions from threads.py (including twist_threads_get) as MCP tools using FastMCP.tool() decorator.

  # Register all tools from tool modules
  for module in [src.inbox, src.threads]:
      for name, func in inspect.getmembers(module, inspect.isfunction):
          if name.startswith('twist_') and func.__module__ == module.__name__:
              logger.info(f"Registering tool: {name}")
              mcp.tool()(func)