A client for the Amazon Selling Partner API
generated from Amazon's API models by an oagen
emitter (codegen/), the way the WorkOS tutorial
How to build a custom SDK generator with oagen
describes: the spec is the source of truth, amzn_selling_partner/sdk/ is the
generated SDK (pydantic v2 models, one resource class per API version, an
async client and a sync one with the same surface, the HTTP layer with its
retry and throttling policy, the exceptions), committed to the repository so
installing the package pulls in no generator. The examples below use the
async client, AsyncSellingPartner; SellingPartner is the synchronous
twin with identical resources and methods (see Sync client). Everything Amazon-specific that is not in the specs
(regions, Login-with-Amazon auth with Restricted Data Tokens, document helpers,
notification models) lives in amzn_selling_partner.plugins.amazon_spapi.
- Bug reports: https://github.com/dbritto-dev/amzn-selling-partner-python/issues
- Migration from 0.1.x: MIGRATION.md
- Updating the bundled specs: docs/UPDATING_SPECS.md
pip install amzn-selling-partner
pip install "amzn-selling-partner[aiohttp]"The package depends on httpx2 and pydantic only; the aiohttp extra adds
the aiohttp transport for DefaultAioHttpClient. Python 3.10 or later.
Create an application in Seller Central and authorize it for the seller; the
client needs the LWA client id, client secret and the seller's refresh token.
They can be passed explicitly or read from the environment
(AMZN_SELLING_PARTNER_CLIENT_ID, AMZN_SELLING_PARTNER_CLIENT_SECRET, AMZN_SELLING_PARTNER_REFRESH_TOKEN, or the old
SELLING_PARTNER_APP_* names).
from amzn_selling_partner import AsyncSellingPartner
client = AsyncSellingPartner(
client_id="amzn1.application-oa2-client....",
client_secret="...",
refresh_token="Atzr|...",
)Access tokens are cached and refreshed once per process (single-flight);
grantless operations use the client_credentials grant with the right scope,
and operations that require a Restricted Data Token obtain one through the
Tokens API automatically. Operations that return PII only on request take an
explicit opt-in:
from amzn_selling_partner import Marketplace
from amzn_selling_partner.plugins.amazon_spapi import with_rdt
orders = await client.orders_v0.list_orders(
marketplace_ids=[Marketplace.US],
created_after="2024-01-01T00:00:00Z",
request_options=with_rdt("buyerInfo", "shippingAddress"),
)A custom token store (for example Redis) is any object with get(key) and
set(key, token): AsyncSellingPartner(token_store=MyStore()).
from amzn_selling_partner import AsyncSellingPartner, Marketplace, Region
AsyncSellingPartner(region=Region.EU)
AsyncSellingPartner(marketplace=Marketplace.DE)
AsyncSellingPartner(region=Region.NA, sandbox=True)Region is NA (the default), EU or FE; a marketplace implies its
region; sandbox=True selects the sandbox endpoint of the region.
Marketplace is a str enum of marketplace ids (Marketplace.US == "ATVPDKIKX0DER")
with .region and .country_code; pass its members wherever an operation
takes marketplace ids (marketplace_ids=[Marketplace.US, Marketplace.CA]).
Every API version is a resource on the client (client.orders_v0,
client.orders_v2026_01_01); client.orders is the newest version. Every
method is a coroutine named after the operation as oagen derives it
(list_orders, get_order, create_feed); path parameters, the request body
and required parameters are positional (keywords work too), optional
parameters are keyword-only, all named after the spec's parameters in
snake_case; request bodies are a model from amzn_selling_partner.sdk.models
(a plain dict with the wire names is accepted too). Every enumerated value in
the specs is a str enum in the same package (OrderOrderStatus.SHIPPED,
FeedProcessingStatus.DONE), usable as arguments and returned in responses;
Marketplace covers the marketplace ids. amzn_selling_partner.sdk.resources.OPERATIONS
maps Amazon's operationIds (getOrders) to the method names.
import asyncio
from amzn_selling_partner import AsyncSellingPartner, Marketplace
from amzn_selling_partner.sdk.models.feeds_v2021_06_30 import CreateFeedSpecification
from amzn_selling_partner.sdk.models.orders_v0 import OrderOrderStatus
async def main() -> None:
async with AsyncSellingPartner() as client:
order = await client.orders_v0.get_order("123-1234567-1234567")
if order.payload.order_status is OrderOrderStatus.SHIPPED:
print("shipped")
item = await client.listings_items.get_listings_item(
seller_id="A1SELLER", sku="MY-SKU", marketplace_ids=[Marketplace.US]
)
await client.feeds.create_feed(
CreateFeedSpecification(
feed_type="POST_PRODUCT_DATA",
marketplace_ids=[Marketplace.US],
input_feed_document_id="...",
)
)
asyncio.run(main())The async client runs on httpx2's own transport; aiohttp is opted into by
passing a client on its transport (the aiohttp extra):
AsyncSellingPartner(http_client=DefaultAioHttpClient()). async with closes
the connection pool, await client.aclose() does the same by hand.
Responses are frozen pydantic models generated from the spec
(amzn_selling_partner.sdk.models.orders_v0.Order); enums are str enums.
Errors raise amzn_selling_partner.APIStatusError subclasses
(RateLimitExceededError, NotFoundError, AuthenticationError, ...) with
.status_code, .body (the decoded error list), .request_id and .response.
SellingPartner is the synchronous client: the same resources, method names,
arguments, models and errors, without await; iter_<method> returns a plain
iterator and with / close() release the pool. Both are generated from the
same plan per operation, so they never drift apart.
from amzn_selling_partner import Marketplace, SellingPartner
with SellingPartner() as client:
order = client.orders_v0.get_order("123-1234567-1234567")
for order in client.orders_v0.iter_list_orders(marketplace_ids=[Marketplace.US]):
...Every option below applies to both clients.
Every paginated operation has an iter_<method> twin that yields the items
of every page, following the API's nextToken (and dropping the other
parameters on the next page where Amazon requires it):
from amzn_selling_partner import Marketplace
from amzn_selling_partner.sdk.models.orders_v0 import OrderOrderStatus
page = await client.orders_v0.list_orders(
marketplace_ids=[Marketplace.US], created_after="2024-01-01T00:00:00Z", order_statuses=[OrderOrderStatus.UNSHIPPED]
)
page.payload.orders
page.payload.next_token
async for order in client.orders_v0.iter_list_orders(
marketplace_ids=[Marketplace.US], created_after="2024-01-01T00:00:00Z", order_statuses=[OrderOrderStatus.UNSHIPPED]
):
...Every page fetch goes through the normal request path (auth, retries, throttling).
RequestOptions(raw=True) returns the decoded JSON (dict/list) without building models:
from amzn_selling_partner import RequestOptions
data = await client.orders_v0.get_order("...", request_options=RequestOptions(raw=True))
data["payload"]["OrderStatus"]The HTTP layer is httpx2: connection pooling, timeouts, proxies, TLS, URL and
query encoding and connection retries are its own (httpx2.HTTPTransport(retries=...)).
On top, each operation has a token bucket seeded from the rate/burst table in
the Amazon documentation, and 408, 429 and 5xx responses are retried with
Retry-After (or Amazon's x-amzn-RateLimit-Limit hint) or exponential
backoff. That policy is generated into sdk/_http.py from
sdkBehavior in codegen/oagen.config.ts. Tune with
AsyncSellingPartner(max_retries=..., throttle=False, timeout=httpx2.Timeout(...))
or per call with request_options=RequestOptions(timeout=5.0, max_retries=0);
client.with_options(max_retries=0) derives a client with some options changed
that shares the connection pool; AMZN_SELLING_PARTNER_TIMEOUT overrides the
default timeout.
content = await client.documents.download_report("amzn1.tortuga.4...")
await client.documents.download_report("amzn1.tortuga.4...", path="report.tsv")
feed = await client.documents.create_feed("POST_PRODUCT_DATA", [Marketplace.US], xml, content_type="text/xml; charset=UTF-8")
message = client.notifications_models.parse(sqs_body)download_report returns the decompressed bytes, or streams to path when
given; create_feed uploads the document and creates the feed in one call;
parse turns an SQS message body into the typed notification model without
any I/O.
The SDK builds its httpx2 client through three factories, exported from the
package: DefaultHttpxClient and DefaultAsyncHttpxClient (httpx2's own
transport, the default) and DefaultAioHttpClient (the aiohttp transport, with
the aiohttp extra). Build one with your options, or any httpx2 client, and
pass it as http_client=; it is used as is (its transport, proxies, event
hooks, auth, default headers and cookies apply; the base URL, timeout and
retries are the SDK's) and is yours to close.
import httpx2
from amzn_selling_partner import AsyncSellingPartner, DefaultAioHttpClient, DefaultAsyncHttpxClient, DefaultHttpxClient, SellingPartner
client = AsyncSellingPartner(http_client=DefaultAsyncHttpxClient(proxy="http://proxy:3128", retries=0))
client = AsyncSellingPartner(http_client=DefaultAioHttpClient(proxy="http://proxy:3128"))
client = AsyncSellingPartner(http_client=httpx2.AsyncClient(event_hooks=hooks))
client = AsyncSellingPartner(transport=httpx2.MockTransport(handler))
client = SellingPartner(http_client=DefaultHttpxClient(limits=httpx2.Limits(max_connections=100), http2=True))transport= is the short form for a transport alone (a MockTransport is how
the tests run without a network), and any httpx2 transport option (limits,
verify, proxy, http2, ...) passed to the client constructor reaches the
default factory.
The emitter is not tied to Amazon: pointed at any OpenAPI 3 document it
produces the same kind of standalone SDK (models, resources, HTTP client,
errors). tests/petstore_sdk is generated from the petstore fixtures and
tests/fixtures/tasks-api.yml is the spec from the oagen tutorial:
cd codegen && npm ci --ignore-scripts && npm run build
npx oagen generate --lang python --spec ../tests/fixtures/tasks-api.yml --namespace TasksClient --output ../tasks_sdktasks_sdk/client.py then has AsyncTasksClient / TasksClient. Amazon's
Swagger 2.0 files go through npm run spec:build first, which writes the one
OpenAPI 3 document the generator runs against, codegen/spec/open-api-spec.yaml
(committed, like spec/open-api-spec.yaml in
workos/openapi-spec). See
docs/UPDATING_SPECS.md for the generator.
git clone --recurse-submodules https://github.com/dbritto-dev/amzn-selling-partner-python
uv sync --extra dev --extra aiohttp
uv run pytest
uv run ty check
uv run pytest benchmarks
uvx nox -s security_test
uv run python -m tests.sandbox
cd codegen && npm ci --ignore-scripts && npm run generate
cd codegen && npm test && npm run typecheckruff is the only linter and formatter, ty the only type checker. Every push
to main releases a minor version unless the branch committed major,
minor or patch to .github/release-bump; the release commit removes the
file again.
ty type-checks the generated code too; the benchmarks assert the
generated method stays within 15 % of an equivalent hand-written httpx2 call; the security
session runs bandit and safety as in CI; the sandbox runner sends every
operation its embedded examples. The last two lines regenerate the SDK after a
spec bump (Node 24) and run the generator's own vitest suite and type check.
codegen/spec/*.yaml, src/amzn_selling_partner/sdk and tests/petstore_sdk
are generated; edit the generator (codegen/src/python), the policy
(codegen/src/policy) or the spec build (codegen/src/spec) instead and commit
the regenerated files (CI fails on drift). See
docs/UPDATING_SPECS.md.