anypoint-sdk 0.0.2

SDK for MuleSoft Anypoint Platform

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 unit tested with unittest, designed for high coverage
• Type checked with mypy
• Linted with Flake8

Features

• Authentication: client credentials flow to obtain a bearer token
• Resources: thin wrappers for core endpoints
  • Organisations, Environments, APIs, Policies, Groups, Contracts, Tiers, Applications
• 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.

Configuration

AnypointClient.from_client_credentials accepts common HTTP options:

AnypointClient.from_client_credentials(
    client_id, client_secret,
    base_url="https://anypoint.mulesoft.com",
    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
)

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
    resources/
      apis.py
      applications.py
      contracts.py
      environments.py
      groups.py
      organizations.py
      policies.py
      tiers.py
    collectors/
      inventory.py
tests/
  ... standard library unittest suite, fast fakes and helpers ...

Run tests and coverage

coverage run -m unittest
coverage report -m

The repo targets 100 percent coverage. The CI job fails on lower coverage.

Lint and type check

flake8 src tests
mypy src tests

GitHub Actions

A workflow is provided at .github/workflows/ci.yml. It runs Flake8 and the unit 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.