anypoint-sdk

An opinionated Python SDK for MuleSoft Anypoint Platform that focuses on safe HTTP access, defensive parsing, and testability. It includes a high level inventory collector that walks organisations and environments to produce a consolidated JSON view of APIs, policies, contracts, tiers, groups, and client applications.

• Python 3.10+
• HTTP client built on requests
• Fully tested with pytest, designed for high coverage (95%+)
• Type checked with mypy
• Linted with black, isort, and pylint

Features

• Authentication: client credentials flow to obtain a bearer token
• Control planes: US, EU and Government Cloud, selected by short name or URL
• Resources: thin wrappers for core endpoints with full CRUD operations
  • Organisations, Environments, APIs, Policies, Groups, Contracts, Tiers, Applications
  • Exchange assets (create, list, delete, download RAML/OAS specifications)
  • Custom policies (policy definitions and implementations)
  • Observability metrics (API request counts with date range support)
• Inventory collector: aggregates live data into a normalised structure suitable for export or further processing
• Filtering: include or exclude by organisation id or name, and by API name, with optional regular expressions
• Resilience: retries for transient HTTP errors, graceful handling of 401 or 403 when an org membership lacks permissions for environment listing
• Logging: pluggable logger interface so you can inject your own logger

Installation

Editable install for development, including dev tools:

pip install -e .[dev]

Production style install:

pip install anypoint-sdk

Quick start

Set your Connected App credentials in the environment:

export ANYPOINT_CLIENT_ID="...your id..."
export ANYPOINT_CLIENT_SECRET="...your secret..."

Create a client and list accessible organisations and environments:

import os
from anypoint_sdk import AnypointClient

with AnypointClient.from_client_credentials(
    client_id=os.environ["ANYPOINT_CLIENT_ID"],
    client_secret=os.environ["ANYPOINT_CLIENT_SECRET"],
) as client:
    orgs = client.organizations.list_accessible()
    envs_by_org = client.environments.list_by_orgs(orgs, skip_unauthorised=True)
    print(f"Organisations: {len(orgs)}")
    for oid, envs in envs_by_org.items():
        names = [e.get("name") for e in envs if isinstance(e, dict)]
        print(oid, names)

Inventory collection

The inventory collector assembles a rich JSON structure for each API instance, including environment automated policies and client contracts. You can allow it to discover scope automatically or pass the exact orgs and envs you want.

import json
import os
from anypoint_sdk import AnypointClient
from anypoint_sdk.collectors.inventory import (
    build_inventory,
    InventoryOptions,
    InventoryFilters,
)

opts = InventoryOptions(
    base_path="./mulesoft_scan_output",
    include_api_policies=True,
    include_environment_policies=True,
)

# Optional filters
filters = InventoryFilters(
    # org_ids=["ef3c3c6f-eb84-4b14-8f5b-b88dda8c82ae"],
    # org_names=["DNF"],
    # org_name_regex=r"^(DNF|mobile)$",
    # api_names=["api-1", "api-2"],
    # api_name_regex=r"^api-",
)

with AnypointClient.from_client_credentials(
    client_id=os.environ["ANYPOINT_CLIENT_ID"],
    client_secret=os.environ["ANYPOINT_CLIENT_SECRET"],
) as client:
    # You may pass orgs and envs explicitly, or omit to let the collector discover them.
    orgs = client.organizations.list_accessible()
    envs_by_org = client.environments.list_by_orgs(orgs, skip_unauthorised=True)

    records = build_inventory(
        client,
        orgs=orgs,                 # or a list of org ids, or omit entirely
        envs_by_org=envs_by_org,   # or omit to let the collector fetch
        options=opts,
        filters=filters,
    )

print(f"Collected {len(records)} API records")
with open("anypoint_inventory.json", "w", encoding="utf-8") as f:
    json.dump(records, f, indent=2)

Example record, trimmed for brevity:

{
  "api_name": "api-1",
  "api_version": "v1",
  "metadata": {
    "source": "anypoint_live_api",
    "anypoint_data": {
      "api_id": "20473240",
      "environment_id": "e2b0de6d-e837-4ae4-9535-f9ee53100e9d",
      "organization_id": "ef3c3c6f-eb84-4b14-8f5b-b88dda8c82ae",
      "client_applications": [ { "app_id": "2693438", "contract_id": "7534804" } ],
      "sla_tiers": [ { "tier_id": "2247207", "scope": "api" } ]
    }
  },
  "policy_configurations": [ { "policy_name": "rate-limiting-sla-based" } ]
}

Scoping rules

• If you pass orgs explicitly, only those organisations are processed. This can be a list of dicts with id and name, or a list of organisation ids. In this case InventoryFilters that pertain to organisations are ignored.
• API filters always apply.

Performance notes

The collector performs per environment and per instance calls as needed. If you already have orgs and envs_by_org from a cached run, pass them back in to reduce calls.

Observability Metrics

The SDK includes support for the MuleSoft Observability API to fetch request metrics.

Get API Request Counts

from datetime import datetime, timedelta

with AnypointClient.from_client_credentials(
    client_id=client_id,
    client_secret=client_secret,
) as client:
    # Get request count for the last 7 days
    count = client.observability.get_api_request_count(
        org_id="your-org-id",
        env_id="your-env-id",
        api_instance_id=20612480,
        start="2025-01-01",
        end="2025-01-07"
    )
    print(f"Total requests: {count:,}")

Date Range Handling

The Observability API has a 30-day maximum per query. The SDK automatically handles larger ranges:

# Large range (automatically split into multiple 30-day queries)
count = client.observability.get_api_request_count(
    org_id="your-org-id",
    env_id="your-env-id",
    api_instance_id=20612480,
    start="2025-01-01",
    end="2025-12-31",  # Full year - automatically split
    auto_split=True  # Default
)

# Disable auto-split (will error if >30 days)
count = client.observability.get_api_request_count(
    org_id="your-org-id",
    env_id="your-env-id",
    api_instance_id=20612480,
    start="2025-01-01",
    end="2025-01-31",
    auto_split=False
)

Flexible Date Input

The observability methods accept multiple date formats:

from datetime import date, datetime, timedelta

# Date strings (most convenient)
count = client.observability.get_api_request_count(
    org_id="org-id",
    env_id="env-id",
    api_instance_id=123,
    start="2025-01-01",
    end="2025-01-31"
)

# Date objects
count = client.observability.get_api_request_count(
    org_id="org-id",
    env_id="env-id",
    api_instance_id=123,
    start=date(2025, 1, 1),
    end=date(2025, 1, 31)
)

# Datetime objects
now = datetime.now()
yesterday = now - timedelta(days=1)
count = client.observability.get_api_request_count(
    org_id="org-id",
    env_id="env-id",
    api_instance_id=123,
    start=yesterday,
    end=now
)

# Unix timestamps in milliseconds
count = client.observability.get_api_request_count(
    org_id="org-id",
    env_id="env-id",
    api_instance_id=123,
    start=1736956800000,
    end=1737043199999
)

Note: End dates are inclusive. "2025-01-31" means up to 23:59:59.999 on that day.

Environment Scanner

The SDK includes a comprehensive environment scanner that combines policies and metrics:

#!/usr/bin/env python3
import os
from datetime import datetime, timedelta
from anypoint_sdk import AnypointClient

# Set environment variables
os.environ["ANYPOINT_ORG_ID"] = "your-org-id"
os.environ["ANYPOINT_ENV_ID"] = "your-env-id"
os.environ["START_DATE"] = "2025-01-01"
os.environ["END_DATE"] = "2025-01-31"

# See examples/scan_environment.py for full implementation

The scanner provides:

• All APIs in the environment
• Request counts per API for a date range
• Environment automated policies
• API instance policies
• Summary statistics and reports

Exchange Assets

The SDK provides full support for Exchange asset management, including RAML and OAS specification downloads.

List and Search Assets

# List all assets
assets = client.exchange.list_assets()

# Search for specific assets
assets = client.exchange.list_assets(org_id="your-org-id", search="customer")

# List policy assets
policies = client.exchange.list_policy_assets(org_id="your-org-id")

Download Specifications

RAML Downloads

# Download RAML bundle (preserves directory structure)
result = client.exchange.download_raml_bundle(
    group_id="your-org-id",
    asset_id="your-api",
    version="1.0.0",
    output_dir="./raml-output"
)
print(f"Main file: {result['main_file']}")
print(f"Extracted files: {len(result['extracted_files'])}")

# Check if RAML is available before downloading
availability = client.exchange.check_raml_available(
    group_id="your-org-id",
    asset_id="your-api",
    version="1.0.0"
)
if availability["available"]:
    print(f"RAML available: {availability['classifiers']}")

OAS (OpenAPI) Downloads

# Download OAS as YAML
client.exchange.download_oas_as_yaml(
    group_id="your-org-id",
    asset_id="your-api",
    version="1.0.0",
    output_path="./api-spec.yaml"
)

# Download OAS as JSON
client.exchange.download_oas_as_json(
    group_id="your-org-id",
    asset_id="your-api",
    version="1.0.0",
    output_path="./api-spec.json"
)

# Check OAS availability
oas_info = client.exchange.check_oas_available(
    group_id="your-org-id",
    asset_id="your-api",
    version="1.0.0"
)

Create Assets with RAML

# Create an Exchange asset with a RAML bundle (ZIP file)
result = client.exchange.create_asset_with_raml_bundle(
    org_id="your-org-id",
    group_id="your-org-id",
    asset_id="my-new-api",
    version="1.0.0",
    raml_zip_path="./my-api-raml.zip",
    name="My New API",
    description="API description",
    main_file="api.raml",
    sync_publication=True
)

Custom Policies

# Create a custom policy (both definition and implementation)
result = client.exchange.create_custom_policy(
    org_id="your-org-id",
    policy_asset_id="my-custom-policy",
    version="1.0.0",
    name="My Custom Policy",
    description="Policy description",
    schema_json='{"type": "object", ...}',
    policy_jar_path="./policy.jar"
)

# Or create definition and implementation separately
client.exchange.create_policy_definition(
    org_id="your-org-id",
    asset_id="my-policy",
    version="1.0.0",
    name="My Policy",
    schema_json=schema_content,
    metadata_yaml=metadata_content
)

client.exchange.create_policy_implementation(
    org_id="your-org-id",
    asset_id="my-policy-imp",
    version="1.0.0",
    name="My Policy",
    policy_jar_path="./policy.jar",
    implementation_yaml=impl_content,
    policy_definition_gav="your-org-id:my-policy:1.0.0"
)

Configuration

AnypointClient.from_client_credentials accepts common HTTP options:

AnypointClient.from_client_credentials(
    client_id, client_secret,
    base_url="https://anypoint.mulesoft.com",  # or "us", "eu", "gov", see Control planes
    timeout=30.0,
    verify=True,                 # set False to skip TLS verification, not recommended
    cert=None,                   # path to client cert bundle if required
    proxies={"https": "http://proxy.example:8080"},
    extra_headers={"X-Requested-By": "scanner"},
    logger=my_logger,            # optional custom logger
)

Control planes

Anypoint has more than one control plane, each at its own address. base_url selects it, on AnypointClient(...) and on AnypointClient.from_client_credentials(...), as a full URL or a short name (case-insensitive):

Short name	Control plane	Base URL	Exchange Maven address
us (default)	US	https://anypoint.mulesoft.com	https://maven.anypoint.mulesoft.com
eu, eu1	EU	https://eu1.anypoint.mulesoft.com	https://maven.eu1.anypoint.mulesoft.com
gov	Government Cloud	https://gov.anypoint.mulesoft.com	https://maven.gov.anypoint.mulesoft.com

with AnypointClient.from_client_credentials(
    client_id, client_secret, base_url="eu"
) as client:
    ...

Login and every resource call go to that address. Exchange asset files (download_asset_file and the RAML and OAS downloads built on it) come from the Maven address of the same control plane, which is the same host prefixed with maven.. A full URL works the same way, so https://anypoint.example.internal downloads from https://maven.anypoint.example.internal. An unknown short name, or a value that is not an http or https URL, raises ValueError.

The resolver and the Maven address helper are public, for tools that take the control plane from their own configuration:

from anypoint_sdk import CONTROL_PLANES, DEFAULT_BASE_URL, maven_base_url, resolve_base_url

resolve_base_url("eu")        # "https://eu1.anypoint.mulesoft.com"
resolve_base_url(None)        # ANYPOINT_BASE_URL if set, otherwise the US address
maven_base_url("eu")          # "https://maven.eu1.anypoint.mulesoft.com"

resolve_base_url falls back to the ANYPOINT_BASE_URL environment variable when it is given no value. AnypointClient itself does not read the environment: without a base_url it uses the US control plane.

Logging

The SDK uses a small logger protocol, so you can pass your own logger that exposes .debug, .info, .warning, .error, and .child(name). If you do not pass one, a sensible default is created with names such as anypoint_sdk.resources.environments and anypoint_sdk.collector.inventory.

Error handling

• HTTP errors raise HttpError(status, message, body). You can catch this around specific calls if you want to continue on 401 or 403 for particular endpoints.
• Network errors are raised as RuntimeError with a short message.
• The inventory collector handles common permission and shape issues internally, it logs at debug or warning level and continues where safe.

Development

Project layout

src/
  anypoint_sdk/
    _http.py
    _logging.py
    _version.py
    auth.py
    client.py
    control_planes.py
    resources/
      apis.py
      applications.py
      contracts.py
      environments.py
      exchange.py
      groups.py
      observability.py
      organizations.py
      policies.py
      tiers.py
    collectors/
      inventory.py
tests/
  ... pytest suite, fast fakes and helpers ...

Run tests and coverage

make test       # Run all tests with coverage
make test-unit  # Run unit tests only

Or directly with pytest:

pytest tests/ --cov=anypoint_sdk --cov-report=term-missing

The repo targets 95% coverage. The CI job fails on lower coverage.

Lint and type check

make lint       # Run black, isort, mypy, and pylint

Or run individual checks:

black --check src tests
isort --check-only src tests
mypy src tests
pylint src

GitHub Actions

A workflow is provided at .github/workflows/ci.yml. It runs linting (black, isort, mypy, pylint) and the test suite with coverage on Python 3.10 to 3.12.

Versioning

The package follows Semantic Versioning. The runtime __version__ is read from installed distribution metadata. During development install in editable mode to expose the version to importlib.metadata.

Changelog

See CHANGELOG.md for notable changes.

Security

• The SDK avoids logging sensitive fields. If you inject a custom logger, review its configuration before enabling debug level in production.

Licence

MIT Licence. See LICENSE.