Twist MCP Server

Instructions

Implementation Reference

• src/threads.py:12-32 (handler)

  The main handler function for the 'twist_threads_getone' tool. It takes a context and thread ID, constructs parameters, and makes a request to the Twist API endpoint 'threads/getone' via twist_request to retrieve the thread data.

  def twist_threads_getone(
      ctx: Context,
      id: int
  ) -> Union[str, Dict[str, Any]]:
      """Gets a thread object by id.
  
      Args:
          id: The id of the thread
      """
      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 thread with ID: {id}")
          thread_data = twist_request("threads/getone", params=params, token=token)
          logger.info(f"Retrieved thread with ID: {id}")
          return thread_data
      except Exception as error:
          logger.error(f"Error getting thread: {error}")
          return f"Error getting thread: {str(error)}"

• src/api.py:28-63 (helper)

  The helper function used by the handler to make HTTP requests to the Twist API. It is called with endpoint='threads/getone' in the handler.

  def twist_request(endpoint, params=None, token=None, method="GET"):
      """
      Make an API request to Twist.
  
      Args:
          endpoint (str): API endpoint to call (without the base URL)
          params (dict, optional): Dictionary of parameters to include in the request
          token (str, optional): Authentication token (if None, uses the one from get_api_client)
          method (str, optional): HTTP method to use (default: "GET")
  
      Returns:
          dict: Response data as a dictionary
  
      Raises:
          Exception: If the API request fails
      """
      if token is None:
          token = get_api_client()
  
      base_url = "https://api.twist.com/api/v3/"
      headers = {"Authorization": f"Bearer {token}"}
      url = f"{base_url}{endpoint}"
  
      try:
          if method == "GET":
              response = requests.get(url, params=params, headers=headers)
          elif method == "POST":
              response = requests.post(url, data=params, headers=headers)
          else:
              raise ValueError(f"Unsupported HTTP method: {method}")
  
          response.raise_for_status()  # Raise an exception for 4XX/5XX responses
          return response.json()
      except requests.exceptions.RequestException as e:
          logger.error(f"Twist API request failed: {e}")
          raise