Integration Client (HTTP Calls)

Modified on: Tue, 14 Jul, 2026 at 11:36 AM

The IntegrationClient is a powerful HTTP client designed for making requests to external APIs in Maltego transforms. It provides built-in rate limiting, concurrency control, and error handling.

Working examples using IntegrationClient are available in the template project. Run ``maltego-transforms start my_project`` and see ``transforms/async_get.py`` to experiment with these patterns directly.

Why Use IntegrationClient?

When building transforms that call external APIs, you need to:

Rate limit requests to avoid hitting API quotas
Control concurrency to prevent overwhelming APIs (or your server)
Handle errors gracefully without crashing transforms
Share connections efficiently across transform executions

IntegrationClient handles all of this out of the box.

Quick Start

This example is from transforms/async_get.py in the template project:

from maltego.server import (
    register_transform, IntegrationClient,
    MaltegoContext, MaltegoServerSettings, run_server
)
from maltego.entities import URL, Phrase

# Create a client at module level (shared across transforms)
client = IntegrationClient()

@register_transform(
    display_name="Get Status Code of URL [New Maltego Integration]",
    description="Fetches a URL and returns its status code",
    transform_set="New Maltego Integration",
)
async def get_status_code(
    input_entity: URL,
    context: MaltegoContext,
) -> Phrase:
    res = await client.get(input_entity.value, context=context)
    return Phrase(
        f"URL {input_entity.value} returned status code: {res.status_code}!"
    )

Configuration

Basic Configuration

from maltego.server import IntegrationClient

client = IntegrationClient(
    # Concurrency control
    max_concurrent=50,              # Max parallel requests globally (default: 50)
    max_concurrent_per_key=6,       # Max parallel requests per user/API key (default: 6)

    # Rate limiting
    max_calls_per_period=25,        # Max requests per period (default: 25)
    period_length_seconds=1.0,      # Period duration in seconds (default: 1.0)

    # Connection settings
    timeout=30,                     # Request timeout in seconds (default: 30)
    verify_ssl=True,                # Verify SSL certificates (default: True)

    # Throttling key strategy
    use_api_key_for_throttling=True,     # Throttle per API key (default: True)
    use_client_ip_for_throttling=False,  # Throttle per client IP (default: False)
)

Rate Limiting Explained

The rate limiter uses a leaky bucket algorithm:

client = IntegrationClient(
    max_calls_per_period=10,      # Allow 10 requests...
    period_length_seconds=1.0,    # ...per second
)
# Effective rate: 10 requests/second

This prevents bursts that could trigger API rate limits.

Concurrency Control Explained

Concurrency is controlled at two levels:

client = IntegrationClient(
    max_concurrent=100,           # Total parallel requests across all users
    max_concurrent_per_key=5,     # Parallel requests per individual user
)

Why both?

max_concurrent protects your server from overload
max_concurrent_per_key ensures fair sharing between users

If one user makes many requests, they won't block others.

Throttling Keys vs External API Keys

``context.api_key`` is NOT the same as your external API's authentication token!


- ``context.rate_limit_key`` = The validated Maltego identity key (automatically used for throttling when ``use_api_key_for_throttling=True``). This is derived from the authenticated JWT/OIDC identity (``context.identity.sub`` / ``context.identity.maltego_id``) and is the recommended forward identifier.


- ``context.api_key`` = The legacy ``Maltego-API-Key`` header value. Still accessible for backwards compatibility. It was always a passthrough value for transform logic, not an auth mechanism -- the server never validated it as a credential. When a validated identity is present, ``context.rate_limit_key`` is populated and takes precedence for throttling.


- External API key = The token for the third-party API you're calling

@register_transform(
    settings=[
        TransformSetting(name="external_api_key", display_name="API Key"),
    ],
)
async def fetch_data(input_entity, settings: Dict[str, Any], context: MaltegoContext):
    external_key = settings.get("external_api_key")  # For authenticating with external API

    # context.rate_limit_key (from the configured auth backend identity) used automatically for throttling.
    # Falls back to context.api_key only when no validated identity is present.
    response = await client.get(
        url="https://api.example.com/data",
        context=context,
        headers={"Authorization": f"Bearer {external_key}"},  # External API auth
    )

    # To identify the current user, prefer context.identity:
    user_id = context.identity.sub if context.identity else None
    maltego_id = context.identity.maltego_id if context.identity else None

By default, throttling happens per context.rate_limit_key (validated identity) automatically, falling back to the legacy context.api_key when no JWT identity is present. You don't need to pass client_identifier unless you want custom grouping (e.g., throttle per organization instead of per user).

Making Requests

GET Requests

@register_transform
async def fetch_data(input_entity: URL, context: MaltegoContext):
    response = await client.get(
        url="https://api.example.com/data",
        context=context,
        params={"query": input_entity.value},
    )

    data = response.json()
    return Phrase(data["result"])

POST Requests

@register_transform
async def create_item(input_entity: Phrase, context: MaltegoContext):
    response = await client.post(
        url="https://api.example.com/items",
        context=context,
        json={"name": input_entity.value},  # JSON body
        headers={"Content-Type": "application/json"},
    )

    return Phrase(f"Created: {response.json()['id']}")

POST with Raw Content

response = await client.post(
    url="https://api.example.com/upload",
    context=context,
    content="raw string content",  # or bytes
    headers={"Content-Type": "text/plain"},
)

PUT, PATCH, DELETE, HEAD

All four methods have the same signature as get()/post() and go through the same rate-limiting and concurrency throttling:

# Update a resource
response = await client.put(
    url="https://api.example.com/items/42",
    context=context,
    json={"name": "updated"},
)

# Partial update
response = await client.patch(
    url="https://api.example.com/items/42",
    context=context,
    json={"status": "active"},
)

# Delete a resource
response = await client.delete(
    url="https://api.example.com/items/42",
    context=context,
)

# Check existence / headers without a body
response = await client.head(
    url="https://api.example.com/items/42",
    context=context,
)

Arbitrary methods with `request()`

Use request() when you need a method not covered by the named helpers. All throttling still applies:

response = await client.request(
    "OPTIONS",
    url="https://api.example.com/items",
    context=context,
)

Streaming Results with AsyncGenerator

For transforms that make multiple requests, use an AsyncGenerator to stream results back to the user as they arrive. This example is from transforms/async_get.py:

from typing import AsyncGenerator, Dict, Any
import asyncio
from maltego.server import register_transform, IntegrationClient, TransformSetting, MaltegoContext
from maltego.entities import URL, Phrase

client = IntegrationClient()

@register_transform(
    display_name="Monitor Status Code of URL [New Maltego Integration]",
    description="Repeatedly checks a URL and streams results",
    transform_set="New Maltego Integration",
    settings=[
        TransformSetting(
            name="steps", display_name="Amount of Queries",
            popup=True, optional=False, type="int", default_value=10
        )
    ],
)
async def monitor_status_code(
    input_entity: URL,
    settings: Dict[str, Any],
    context: MaltegoContext,
) -> AsyncGenerator[Phrase, None]:
    steps = settings.get("steps")
    if not steps:
        return

    for i in range(steps):
        res = await client.get(input_entity.value, context=context)
        context.log.inform(f"Query step {i} returned {res.status_code}")
        yield Phrase(
            f"URL {input_entity.value} returned status code: {res.status_code} ({i})!"
        )
        await asyncio.sleep(1)

Benefits of streaming:

Users see results as they arrive
Long-running transforms show progress
Memory efficient for large result sets

Error Handling

The client raises specific exceptions for different error types:

from maltego.model.exception import (
    MaltegoException,
    MaltegoHTTPDataProviderAPIKeyInvalid,  # 401 Unauthorized
    MaltegoHTTPUnauthorized,               # 403 Forbidden
    MaltegoHTTPDataProviderNotFound,       # 404 Not Found
    MaltegoHTTPDataProviderUnavailable,    # 5xx Server Errors
)

@register_transform
async def safe_fetch(input_entity: URL, context: MaltegoContext):
    try:
        response = await client.get(input_entity.value, context=context)
        return Phrase(response.text)

    except MaltegoHTTPDataProviderAPIKeyInvalid:
        context.log.fatal("Invalid API key - check your credentials")
        return None

    except MaltegoHTTPDataProviderNotFound:
        context.log.inform(f"Resource not found: {input_entity.value}")
        return None

    except MaltegoHTTPDataProviderUnavailable:
        context.log.fatal("API is currently unavailable, try again later")
        return None

    except MaltegoException as e:
        context.log.fatal(f"Request failed: {e}")
        return None

Response Hooks

Response hooks let you inspect or modify every response:

from httpx import Response
from maltego.model.context import MaltegoContext

def log_response(response: Response, context: MaltegoContext, **kwargs):
    """Log every API response for debugging."""
    context.log.debug(f"API returned {response.status_code} for {response.url}")

def check_rate_limit(response: Response, context: MaltegoContext, **kwargs):
    """Inform user when approaching rate limits."""
    remaining = response.headers.get("X-RateLimit-Remaining")
    if remaining and int(remaining) < 10:
        context.log.partial(f"Rate limit warning: {remaining} requests remaining")

client = IntegrationClient(
    response_hooks=[log_response, check_rate_limit],
)

Response hooks must be synchronous functions. Async hooks are not supported. Raising a ``MaltegoException`` subclass inside a hook will propagate to the transform caller as-is; any other exception is wrapped in a generic ``MaltegoException``.

Direct `httpx_client` Access

IntegrationClient exposes the underlying httpx.AsyncClient as client.httpx_client for advanced use. Calls made directly through httpx_client bypass SDK rate limiting, concurrency throttling, and Maltego exception mapping. Only use direct access when those protections are intentionally not needed, and prefer the public request methods in all other cases.

Advanced: Custom Concurrency Limits

Override concurrency limits for specific users or contexts:

@register_transform
async def premium_user_fetch(input_entity: URL, context: MaltegoContext):
    # Give premium users higher concurrency
    if is_premium_user(context.identity):
        client.override_concurrency_limit_for(
            context=context,
            new_limit=20,  # 20 parallel requests instead of 6
        )

    response = await client.get(input_entity.value, context=context)
    return Phrase(response.text)

Advanced: Proxy Support

The proxies parameter accepts a URL-keyed dict mapping scheme/host prefixes to proxy URLs. The SDK adapts the underlying httpx.AsyncClient constructor argument name (proxy vs proxies) for compatibility with the installed httpx version at runtime; the value you provide is passed through unchanged.

client = IntegrationClient(
    proxies={
        "http://": "http://proxy.example.com:8080",
        "https://": "http://proxy.example.com:8080",
    },
    trust_env=True,  # Also respect HTTP_PROXY / HTTPS_PROXY env vars
)

``trust_env=True`` (the default) instructs ``httpx`` to read ``HTTP_PROXY``, ``HTTPS_PROXY``, and ``ALL_PROXY`` environment variables. Set ``trust_env=False`` to ignore environment proxies entirely.

Warning: In the current release, ``proxies`` and ``trust_env`` are not preserved when the underlying HTTP client is internally reset after a connection failure. If proxy routing is critical, recreate the ``IntegrationClient`` after a ``MaltegoHTTPDataProviderUnavailable`` exception instead of reusing the same instance. This limitation will be addressed in a future lifecycle-management release.

Integration with Pagination

The IntegrationClient integrates seamlessly with the pagination module:

from maltego.pagination import OffsetLimitPaginator

client = IntegrationClient(
    max_calls_per_period=10,  # Respect API rate limits
)

paginator = OffsetLimitPaginator(
    client=client,  # Paginator uses the client for all requests
    response_to_items=parse_items,
    page_size=50,
)

# All paginated requests are rate-limited and throttled
items = await paginator.fetch_all_items(slider=500, context=context, url=api_url)

See the Pagination article for details.

Best Practices

1. Create Clients at Module Level

# Good: Shared client, connections reused
client = IntegrationClient()

@register_transform
async def transform1(entity, context):
    await client.get(...)

@register_transform
async def transform2(entity, context):
    await client.get(...)  # Reuses same client

# Bad: New client per request, no connection reuse
@register_transform
async def transform(entity, context):
    client = IntegrationClient()  # Created on every call!
    await client.get(...)

Threaded runner caveats: The default server runner (``transform_runner=THREADED``) starts with a single worker thread and one asyncio event loop, so module-level clients are safe to share in the default configuration. If you increase ``num_worker`` in ``MaltegoServerSettings``, each worker thread runs its own event loop and a module-level ``IntegrationClient``'s asyncio internals (semaphores, rate-limit tasks) are bound to the first loop that initialises them. In multi-worker configurations, create one ``IntegrationClient`` per worker or keep ``num_worker=1`` (the default) to avoid cross-loop sharing.

2. Configure Rate Limits for Your API

Check your API's documentation and configure accordingly:

# Example: API allows 100 requests per minute
client = IntegrationClient(
    max_calls_per_period=100,
    period_length_seconds=60.0,
)

# Example: API allows 5 requests per second
client = IntegrationClient(
    max_calls_per_period=5,
    period_length_seconds=1.0,
)

3. Use Context for Logging

Always pass context and use context.log for user feedback:

@register_transform
async def fetch(entity, context: MaltegoContext):
    context.log.inform("Fetching data...")

    response = await client.get(url, context=context)

    context.log.inform(f"Retrieved {len(response.json())} items")
    return results

4. Handle Partial Failures Gracefully

When fetching multiple items, don't let one failure break everything:

@register_transform
async def fetch_multiple(entities: List[URL], context: MaltegoContext):
    results = []

    for entity in entities:
        try:
            response = await client.get(entity.value, context=context)
            results.append(Phrase(response.json()["title"]))
        except MaltegoException:
            context.log.partial(f"Failed to fetch {entity.value}")
            # Continue with other entities

    return results

Client Lifecycle

For most use cases, creating a module-level IntegrationClient and never closing it works fine -- the server process owns the client for its entire lifetime.

If you need explicit cleanup (e.g., in tests or short-lived workers), call aclose():

await client.aclose()

aclose() is idempotent -- calling it multiple times is safe.

Async context manager

IntegrationClient supports async with, which calls aclose() automatically when the block exits:

async with IntegrationClient() as client:
    response = await client.get(url, context=context)
    # client.aclose() is called here even if an exception is raised

Calls to a closed client will fail. Do not reuse a client after ``aclose()`` or after the ``async with`` block exits.

Try It Yourself

To experiment with IntegrationClient:

# Create a new project from the template
maltego-transforms start my_project
cd my_project

# The IntegrationClient examples are in:
# transforms/async_get.py

# Start the server
python project.py

Then configure your Maltego client with the seed URL and run the transforms on a URL entity.

API Reference

IntegrationClient (maltego.util.IntegrationClient) is the client used to interact with external APIs providing data for Maltego transforms.

Constructor (__init__) parameters:

Parameter	Type	Default	Description
`max_concurrent`	int	`50`	Max parallel requests globally.
`max_concurrent_per_key`	int	`6`	Max parallel requests per user/API key.
`max_calls_per_period`	int	`25`	Max requests per rate-limit period (leaky bucket).
`period_length_seconds`	float	`1.0`	Length of the rate-limit period, in seconds.
`timeout`	int	`30`	Request timeout, in seconds.
`verify_ssl`	bool	`True`	Whether to verify SSL certificates.
`use_api_key_for_throttling`	bool	`True`	Throttle per validated Maltego identity (`context.rate_limit_key`), falling back to the legacy `context.api_key` when no identity is present.
`use_client_ip_for_throttling`	bool	`False`	Throttle per client IP instead.
`proxies`	Optional	`None`	URL-keyed dict of scheme/host prefixes to proxy URLs.
`trust_env`	bool	`True`	Whether the underlying `httpx` client reads `HTTP_PROXY`/`HTTPS_PROXY`/`ALL_PROXY` env vars.
`response_hooks`	Optional	`None`	List of synchronous callables invoked with each response for inspection/logging.

Key methods:

Method	Description
`get(url, context, headers=None, params=None, client_identifier=None, **kwargs)`	Performs a throttled/rate-limited GET request and returns an `httpx.Response`.
`post(url, context, content=None, json=None, headers=None, params=None, client_identifier=None, **kwargs)`	Performs a throttled/rate-limited POST request (JSON body or raw `content`) and returns an `httpx.Response`.
`put(...)`, `patch(...)`, `delete(...)`, `head(...)`	Same signature and throttling as `get()`/`post()`, for updating, partially updating, deleting, or checking a resource.
`request(method, url, context, **kwargs)`	Performs a throttled/rate-limited request using an arbitrary HTTP method not covered by the named helpers.
`aclose()`	Closes the underlying HTTP client. Idempotent. Also available as an async context manager (`async with IntegrationClient() as client:`), which calls `aclose()` automatically.
`override_concurrency_limit_for(context, new_limit, client_identifier=None)`	Overrides the per-user/context concurrency limit at runtime. `client_identifier` optionally sets a custom throttling key (instead of the default derived from `context`) so the override applies to a custom grouping, e.g. per organization instead of per user.
`httpx_client` (attribute)	The underlying `httpx.AsyncClient` instance for advanced/direct access, bypassing SDK throttling and exception mapping.

For the full API reference, see the SDK API Reference article.