async-amazon-ads-api-v1

Pure async Amazon Ads API v1 client — Sponsored Products, Sponsored Brands, Sponsored Display.

Installation

pip install async-amazon-ads-api-v1
# 或
uv add async-amazon-ads-api-v1

Redis 缓存支持：

pip install "async-amazon-ads-api-v1[redis]"
# 或
uv add "async-amazon-ads-api-v1[redis]"

Quick Start

所有 API 方法仅接受 Pydantic model 实例，不支持 dict。

使用 Access Token（直接提供）

import asyncio
from async_amazon_ads_api_v1 import AmazonAdsConfig, Region, SPClient
from async_amazon_ads_api_v1.models.sp import SPQueryCampaignRequest


async def main() -> None:
    config = AmazonAdsConfig(
        access_token="your-access-token",
        client_id="your-client-id",
        region=Region.NA,
    )
    body = SPQueryCampaignRequest(
        adProductFilter={"include": ["SPONSORED_PRODUCTS"]},
        stateFilter={"include": ["ENABLED"]},
    )

    async with SPClient(config) as sp:
        resp = await sp.campaigns.query(body)
        print(resp.model_dump_json(indent=2))


asyncio.run(main())

使用 Refresh Token（自动续期）

import asyncio
from async_amazon_ads_api_v1 import AmazonAdsConfig, Region, SPClient
from async_amazon_ads_api_v1.models.sp import SPQueryCampaignRequest


async def main() -> None:
    config = AmazonAdsConfig(
        access_token="your-access-token",   # 可选，提供后直接使用
        client_id="your-client-id",
        refresh_token="your-refresh-token", # access_token 过期后自动刷新
        client_secret="your-client-secret",
        region=Region.NA,
    )
    body = SPQueryCampaignRequest(
        adProductFilter={"include": ["SPONSORED_PRODUCTS"]},
        stateFilter={"include": ["ENABLED"]},
    )

    async with SPClient(config) as sp:
        resp = await sp.campaigns.query(body)
        print(resp.model_dump_json(indent=2))


asyncio.run(main())

从环境变量加载

export AMAZON_ACCESS_TOKEN=...
export AMAZON_CLIENT_ID=...
export AMAZON_REFRESH_TOKEN=...     # 可选，用于自动续期
export AMAZON_CLIENT_SECRET=...     # 可选，refresh_token 时需要
export AMAZON_REGION=na             # na | eu | fe，默认 na
export AMAZON_PROFILE_ID=...        # 可选

from async_amazon_ads_api_v1.config.loader import from_toml

config = from_toml()

Environment Variables

Token Management

SDK 内置 OAuth token 生命周期管理，支持自动续期和缓存：

from async_amazon_ads_api_v1 import AmazonAdsConfig, Region

# 自动续期 + 文件缓存（默认）
config = AmazonAdsConfig(
    client_id="your-client-id",
    client_secret="your-client-secret",
    refresh_token="your-refresh-token",
    region=Region.NA,
)

# 自动续期 + Redis 缓存
config = AmazonAdsConfig(
    client_id="your-client-id",
    client_secret="your-client-secret",
    refresh_token="your-refresh-token",
    region=Region.NA,
    cache_backend="redis",
    redis_url="redis://localhost:6379",
)

API Reference

SPClient — Sponsored Products

SBClient — Sponsored Brands

SDClient — Sponsored Display

Legacy / 非标准 API

以下资源类不遵循 /adsApi/v1 通用 CRUD 模式，需要单独实例化：

BudgetRules — 预算规则关联

SP、SB、SD 各自独立的资源类：

from async_amazon_ads_api_v1.client.legacy import SPBudgetRules, SBBudgetRules, SDBudgetRules

sp_rules = SPBudgetRules(ctx)
await sp_rules.create_budget_rules(request)
await sp_rules.update_budget_rules(request)

Portfolios — 投资组合

from async_amazon_ads_api_v1.client.legacy import Portfolios
from async_amazon_ads_api_v1.models.legacy.portfolios import UpdatePortfoliosRequestContent

pf = Portfolios(ctx)
resp = await pf.list()
await pf.update(UpdatePortfoliosRequestContent(portfolios=[...]))

SBOptimizationRules — SB 优化规则 (Beta)

from async_amazon_ads_api_v1.client.legacy import SBOptimizationRules
from async_amazon_ads_api_v1.models.legacy import SBEntityFilter, SBListOptimizationRulesRequest

rules = SBOptimizationRules(ctx)
request = SBListOptimizationRulesRequest(
    entityFilter=SBEntityFilter(entityType="CAMPAIGN", entityId="..."),
)
await rules.list_optimization_rules(request)
await rules.disassociate_optimization_rules(request)

SDOptimizationRules — SD 优化规则 (Beta)

from async_amazon_ads_api_v1.client.legacy import SDOptimizationRules
from async_amazon_ads_api_v1.models.legacy.sd_rules import (
    SDCreateAssociatedOptimizationRulesRequest,
    SDCreateOptimizationRule,
)

rules = SDOptimizationRules(ctx)
await rules.list_optimization_rules(count=10, state_filter="enabled")
await rules.get_optimization_rule("rule-id")
await rules.list_ad_group_optimization_rules(ad_group_id=12345)
await rules.create_optimization_rules([SDCreateOptimizationRule(...)])
await rules.associate_optimization_rules(
    12345,
    SDCreateAssociatedOptimizationRulesRequest(optimizationRuleIds=["rule-id"]),
)
await rules.disassociate_optimization_rules(
    12345,
    SDCreateAssociatedOptimizationRulesRequest(optimizationRuleIds=["rule-id"]),
)

SDCreatives — SD 创意

from async_amazon_ads_api_v1.client.legacy import SDCreatives

creatives = SDCreatives(ctx)
await creatives.create([{...}])

License

MIT