FXMacroData for Prefect

Bring your FXMacroData subscription into scheduled Prefect research workflows: cross-currency indicators, complete available histories, FX reference rates, commodities, positioning and release analysis. Store your API key in a Prefect Credentials Block and pass its name to native tasks.

Subscribe to FXMacroData for subscriber coverage and deeper available history. Public USD catalogue, recent indicator history and calendar access let you evaluate the workflow before subscribing without an API key. Availability varies by dataset.

Install and connect

Use Python 3.10–3.14 and Prefect 3.7.4 or later. From the release artifacts directory, install both supplied wheels into the environment that runs your flows:

python -m pip install ./fxmacrodata_public_client-0.1.1-py3-none-any.whl ./prefect_fxmacrodata-0.1.0-py3-none-any.whl
prefect block register -m prefect_fxmacrodata

The commands use the supplied artifacts rather than assume a package-registry release. Configure your Prefect server or Cloud workspace through Prefect's normal setup. In Blocks, create an FXMacroData Credentials block, enter your subscription API key and save it as fxmacrodata-research. The key uses Prefect's secret-field storage. The timeout controls each HTTP request, in seconds.

To create the same block interactively without putting a key in source code:

from getpass import getpass
from pydantic import SecretStr
from prefect_fxmacrodata import FXMacroDataCredentials

FXMacroDataCredentials(
    api_key=SecretStr(getpass("FXMacroData API key: ")),
).save("fxmacrodata-research")

Create the block during setup, outside a flow. Flows and tasks take only its name; do not put keys in flow parameters, operation arguments or logs. Omitting credentials_block explicitly uses public evaluation access, even if an API-key environment variable is set.

Build a research flow

macro_snapshot is a Prefect subflow that collects every available history page for the selected series and retrieves its release calendar. Pass the typed snapshot to your own downstream task:

import asyncio
from prefect import flow, task
from prefect_fxmacrodata import MacroSnapshot, macro_snapshot


@task(persist_result=False)
def count_observations(snapshot: MacroSnapshot) -> int:
    return len(snapshot.history.records())


@flow(persist_result=False)
async def research_run():
    snapshot = await macro_snapshot(
        "EUR",
        "policy_rate",
        credentials_block="fxmacrodata-research",
        start_date="2020-01-01",
        end_date="2026-01-01",
    )
    return count_observations(snapshot)


if __name__ == "__main__":
    asyncio.run(research_run())

To evaluate with USD, omit the credentials and historical date range:

snapshot = await macro_snapshot("USD", "policy_rate")

Subscriber history can extend beyond the public evaluation window where the series supports it. Deploy or schedule this flow with your normal Prefect work pool and deployment process. The date filters select indicator history. The calendar uses the API's current release-calendar window for the currency; it is not filtered to the historical range or one indicator. Use the release_calendar task directly for different calendar filters.

Use individual operations

Every packaged operation is a named Prefect task, with normal task states, dependencies, with_options, retries and scheduling through its enclosing flow. Both REST and hosted MCP operations are available:

from prefect_fxmacrodata import operation_schemas
from prefect_fxmacrodata.tasks import indicator_history, mcp_macro_briefing_task


async def inspect_series():
    page = await indicator_history(
        {"currency": "GBP", "indicator": "inflation", "limit": 100},
        credentials_block="fxmacrodata-research",
    )
    briefing = await mcp_macro_briefing_task(
        {"currency": "GBP"},
        credentials_block="fxmacrodata-research",
    )
    return page, briefing


schemas = {operation.name: operation.input_schema for operation in operation_schemas()}

See CAPABILITIES.md for the exact operation inventory. The original input schemas are available through operation_schemas() or the native discover_operations task. OPERATION_TASKS[name] supports configuration-driven flows without losing operation-specific task names.

Results are FXMacroDataResult models. payload retains the entire service response, including original units, provenance, unavailable values and MCP content; source_url records the source endpoint. records() adds a convenient row view without replacing the payload. No publication timestamps or missing observations are inferred.

History and failure behavior

Individual operation tasks return a single response using the API's pagination rules. Use complete_history("indicator_history", arguments, credentials_block) for a complete available dataset within the requested filters. It starts at offset zero, verifies contiguous pages and dataset versions when supplied, and fails if completion cannot be established. Its payload retains every page's metadata and returns rows in chronological order. max_pages is an explicit safety budget; an error never passes an incomplete history off as complete. Applications that need resumable storage can persist individual page responses and control offset and dataset_version through the named operation tasks.

Event streaming is a bounded capture using max_events and max_seconds. The response retains the original completion flag and reason. Cancellation finishes the current bounded HTTP request, closes the client, and stops before the next history page. MCP sessions are closed after each task.

Tasks disable result persistence and caching by default. Opt in only when the destination and access controls suit your data licence. Task errors omit raw responses, authenticated URLs and credentials. Transient retries are opt-in with Prefect's with_options(retries=..., retry_delay_seconds=...).

FXMacroData API reference describes access requirements and supported arguments for each dataset.