Implementation:Mage ai Mage ai Intercom Client
| Knowledge Sources | |
|---|---|
| Domains | Data_Integration, Intercom, API |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete HTTP client for making authenticated requests to the Intercom API v2.0, provided by the Mage Intercom source connector.
Description
The IntercomClient class manages all HTTP communication with the Intercom API at https://api.intercom.io. It authenticates using Bearer token via access_token. The client is implemented as a context manager (__enter__/__exit__), verifying the access token on entry by hitting the /tags endpoint and closing the session on exit. Rate limiting is enforced via the Singer @utils.ratelimit(1000, 60) decorator on both check_access_token and request methods. The client sets Intercom-Version: 2.0 on all requests. It uses Singer metrics.http_request_timer for request timing instrumentation. A comprehensive exception hierarchy maps HTTP status codes (400-500) to specific error classes. Configurable request timeout defaults to 300 seconds.
Usage
Used internally by Intercom stream classes. Requires access_token and optionally request_timeout and user_agent.
Code Reference
Source Location
- Repository: mage-ai
- File:
mage_integrations/mage_integrations/sources/intercom/client.py - Lines: 1-247
Signature
class IntercomClient(object):
def __init__(self,
access_token,
config_request_timeout,
user_agent=None):
Import
from mage_integrations.sources.intercom.client import IntercomClient
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| access_token | str | Yes | Bearer token for Intercom API authentication |
| config_request_timeout | float/str | Yes | Request timeout in seconds (defaults to 300 if falsy) |
| user_agent | str | No | Optional User-Agent header value |
Outputs
| Name | Type | Description |
|---|---|---|
| response | dict | Parsed JSON response from the Intercom API |
Key Methods
| Method | Description |
|---|---|
check_access_token() |
Validates the access token by querying /tags. Rate limited to 1000 req/60s.
|
request(method, path, url, **kwargs) |
Core request method with rate limiting, metric timing, and error handling. Sets Intercom-Version header. |
get(path, **kwargs) |
Convenience wrapper for GET requests. |
post(path, **kwargs) |
Convenience wrapper for POST requests. |
perform(method, path, **kwargs) |
Dispatches to get or post based on method string.
|
Error Handling
Defines a hierarchy of custom exceptions:
| Status Code | Exception Class |
|---|---|
| 400 | IntercomBadRequestError
|
| 401 | IntercomUnauthorizedError
|
| 402 | IntercomPaymentRequiredError
|
| 403 | IntercomForbiddenError
|
| 404 | IntercomNotFoundError
|
| 405 | IntercomMethodNotAllowedError
|
| 406 | IntercomNotAcceptableError
|
| 408 | IntercomRequestTimeoutError
|
| 409 | IntercomUserConflictError
|
| 415 | IntercomUnsupportedMediaTypeError
|
| 422 | IntercomUnprocessableEntityError
|
| 423 | IntercomScrollExistsError
|
| 500 | IntercomInternalServiceError
|
| >=500 | Server5xxError
|
Usage Examples
from mage_integrations.sources.intercom.client import IntercomClient
with IntercomClient(
access_token='your_intercom_access_token',
config_request_timeout=300,
user_agent='MageIntegration/1.0',
) as client:
tags = client.get('tags')
users = client.get('contacts', params={'per_page': 50})