Twist MCP Server

Instructions

Implementation Reference

• src/threads.py:295-321 (handler)

  Handler function that retrieves unread threads for the current workspace using the Twist API. It fetches the workspace ID from environment, makes the API call via twist_request, handles errors, and returns the list of unread threads or error message.

  def twist_threads_get_unread(
      ctx: Context
  ) -> Union[str, List[Dict[str, Any]]]:
      """Gets unread threads in a workspace for the authenticated user.
      """
      token = ctx.request_context.lifespan_context.twist_token
  
      workspace_id = os.getenv("TWIST_WORKSPACE_ID")
      if not workspace_id:
          logger.error("TWIST_WORKSPACE_ID environment variable is required")
          return "Error: TWIST_WORKSPACE_ID environment variable is required"
  
      params = {"workspace_id": workspace_id}
  
      try:
          logger.info(f"Getting unread threads for workspace ID: {workspace_id}")
          unread_data = twist_request("threads/get_unread", params=params, token=token)
  
          if not unread_data:
              logger.info("No unread threads found")
              return "No unread threads found"
  
          logger.info(f"Retrieved {len(unread_data)} unread threads")
          return unread_data
      except Exception as error:
          logger.error(f"Error getting unread threads: {error}")
          return f"Error getting unread threads: {str(error)}"

• main.py:42-47 (registration)

  Dynamic registration of all functions starting with 'twist_' from src.inbox and src.threads modules as MCP tools using FastMCP. This includes the twist_threads_get_unread handler.

  # 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)