Twist MCP Server

Instructions

Implementation Reference

• src/inbox.py:12-66 (handler)

  The main handler function for the 'twist_inbox_get' tool. It fetches the user's inbox threads from the Twist API, applying optional filters like limit, timestamps, archive status, order, and exclusions. Returns JSON string of threads or error message.
  def twist_inbox_get( ctx: Context, limit: Optional[int] = None, newer_than_ts: Optional[int] = None, older_than_ts: Optional[int] = None, archive_filter: Optional[str] = None, order_by: Optional[str] = None, exclude_thread_ids: Optional[List[int]] = None ) -> str: """Get the authenticated user's inbox. Args: limit: Limits the number of threads returned (default is 30, 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 archive_filter: Filter threads based on their is_archived flag: 'all', 'archived', or 'active' (default) order_by: Order of threads: 'desc' (default) or 'asc', based on last_updated attribute exclude_thread_ids: Thread IDs to exclude from results """ 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} if limit is not None: params["limit"] = limit if newer_than_ts is not None: params["newer_than_ts"] = newer_than_ts if older_than_ts is not None: params["older_than_ts"] = older_than_ts if archive_filter is not None: params["archive_filter"] = archive_filter if order_by is not None: params["order_by"] = order_by if exclude_thread_ids is not None: params["exclude_thread_ids"] = exclude_thread_ids try: logger.info(f"Getting inbox for workspace ID: {workspace_id}") inbox_data = twist_request("inbox/get", params=params, token=token) if not inbox_data: logger.info("No inbox threads found") return "No inbox threads found" logger.info(f"Retrieved {len(inbox_data)} inbox threads") return inbox_data except Exception as error: logger.error(f"Error getting inbox: {error}") return f"Error getting inbox: {str(error)}"
• main.py:43-47 (registration)

  Dynamic registration code that discovers and registers all functions starting with 'twist_' from the src.inbox module (including twist_inbox_get) as MCP tools using FastMCP's @tool decorator.
  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)
• src/api.py:28-64 (helper)

  Helper function used by the handler to make authenticated HTTP requests to the Twist API, specifically called with 'inbox/get' endpoint in twist_inbox_get.
  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