Python SDK Migration Guide

The Python SDK provides an idiomatic, async-first interface for integrating with Agent Offer Network. It mirrors the TypeScript SDK surface while following Python dataclass and snake_case conventions.

Migration examples

This guide retains older source-compatible user_profile and pagination examples. New integrations must use the Python SDK Reference and its canonical Protocol v1.0 request shape; those legacy inputs are not emitted on the current wire.

Runtime baseline

The current Python SDK package is agentoffernetwork and requires Python 3.11+. If you are setting up a new project, prefer uv add agentoffernetwork or pip install agentoffernetwork.

Prerequisites

Make sure you have completed the Quick Start and understand the Core Concepts before proceeding.

Installation

pip install agentoffernetwork

Or with uv:

uv add agentoffernetwork

Initialization

The SDK uses an async factory function initialize() that returns an AgentOfferClient instance:

Full Configuration

Python SDK initialization

Python

import os
from agentoffernetwork import initialize, SDKConfig

client = await initialize(SDKConfig(
    api_key=os.environ["AON_API_KEY"],
    mode="live",                  # "live" | "mock"
    base_url="https://api.aon.pro",      # optional, defaults to this environment
    timeout=5.0,                  # optional, seconds (positive number)
))

Configuration Options

Parameter

Description

api_key

strRequired

Your API key for authentication. Never hardcode -- use environment variables.

mode

Literal['live', 'mock']Required

live: real API calls. mock: local mock data for development and testing.

base_url

strOptional

API base URL. Defaults to https://api.aon.pro.

timeout

float | NoneOptional

Request timeout in seconds. Must be positive. Defaults to None (SDK default).

Environment Variable Best Practices

# .env
AON_API_KEY=your-api-key-here
AON_MODE=live

import os
from agentoffernetwork import initialize, SDKConfig
 
client = await initialize(SDKConfig(
    api_key=os.environ["AON_API_KEY"],
    mode=os.environ.get("AON_MODE", "mock"),
))

API Key security

Never commit API keys to version control. Use environment variables or a secret manager. See Best Practices for more on security.

Async/Await Pattern

All network-calling methods in the Python SDK are async. You must use await when calling them:

import asyncio
from agentoffernetwork import (
    initialize, SDKConfig, QueryOffersParams, Intent,
    TextContentPart, QueryContext, UserProfile, DeviceInfo,
)
 
async def main():
    client = await initialize(SDKConfig(api_key="your-key", mode="mock"))
 
    response = await client.query_offers(QueryOffersParams(
        intent=Intent(content=[TextContentPart(type="input_text", text="headphones")]),
        context=QueryContext(user_profile=UserProfile(language="en", device_info=DeviceInfo())),
    ))
 
    for offer in response.offers:
        info = offer.offer_info
        price = info.commercial.price if info.commercial else None
        price_str = f"{price.amount} {price.currency}" if price else "N/A"
        print(f"{info.title} - {price_str}")
 
asyncio.run(main())

Sync exception

The format_recommendation() function is synchronous -- it performs string formatting only and does not make network calls. No await needed.

Querying Offers

The query_offers method accepts QueryOffersParams and returns QueryOffersResponse.

Basic Query

from agentoffernetwork import QueryOffersParams, Intent, TextContentPart, QueryContext, UserProfile, DeviceInfo
 
response = await client.query_offers(QueryOffersParams(
    intent=Intent(content=[TextContentPart(type="input_text", text="project management tools")]),
    context=QueryContext(user_profile=UserProfile(language="en", device_info=DeviceInfo())),
))
 
print(f"Request ID: {response.request_id}")
print(f"Found {len(response.offers)} offers")

Using detect_context

The detect_context helper auto-detects platform, language, and session from environment signals:

from agentoffernetwork import detect_context
 
context = detect_context(
    interests=["project management", "team collaboration"],
)
 
response = await client.query_offers(QueryOffersParams(
    intent=Intent(content=[TextContentPart(type="input_text", text="best PM tools")]),
    context=context,
))

Using create_context_for_platform

Use create_context_for_platform when the host platform is already known and you want stable platform attribution instead of runtime inference.

from agentoffernetwork import create_context_for_platform
 
mcp_context = create_context_for_platform(
    "mcp-skill",
    interests=["noise cancelling headphones"],
)

Advanced Constraints

from agentoffernetwork import QueryConstraints, QueryPagination, DeviceInfo
 
response = await client.query_offers(QueryOffersParams(
    intent=Intent(content=[TextContentPart(type="input_text", text="affordable SaaS tools")]),
    context=QueryContext(user_profile=UserProfile(language="en", device_info=DeviceInfo())),
    constraints=QueryConstraints(
        category_ids=["computers_electronics.computers.software"],
    ),
    pagination=QueryPagination(limit=10, offset=0),
))

Constraint Parameters

Parameter

Description

category_ids

list[CategoryId] | NoneOptional

Constrain by AON Taxonomy v1 category id. Parent ids match descendant offer categories.

Pagination

offset = 0
limit = 20
has_more = True  # SDK-derived convenience, not a protocol response field.
 
while has_more:
    response = await client.query_offers(QueryOffersParams(
        intent=Intent(content=[TextContentPart(type="input_text", text="headphones")]),
        context=QueryContext(user_profile=UserProfile(language="en", device_info=DeviceInfo())),
        pagination=QueryPagination(limit=limit, offset=offset),
    ))
 
    for offer in response.offers:
        print(offer.offer_info.title)
 
    has_more = response.has_more  # SDK-derived convenience.
    offset += limit

Python Type System

The Python SDK uses dataclass types for all models and Literal types for enums. This provides full IDE autocompletion and type checking with mypy or pyright:

from agentoffernetwork import (
    Offer, OfferInfo, Entity, Category, CommercialInfo, Price,
    CategoryId, BidModel, OfferStatus, OfferType, AuditStatus,
)
 
# All fields are typed (Offer is nested per the current protocol)
offer: Offer = response.offers[0]
entity_name: str = offer.entity.name
category_id: CategoryId = offer.offer_info.category.id
commercial: CommercialInfo | None = offer.offer_info.commercial
price: Price | None = commercial.price if commercial else None
 
# Type narrowing
if price is not None:
    # Price.amount is a decimal string (e.g. "349.99"); build display yourself or use SDK helper
    print(f"Price: {price.amount} {price.currency}")

Protocol Interface

The AgentOfferClient is defined as a Protocol (structural typing), enabling duck-typing and easy mocking:

from agentoffernetwork import AgentOfferClient
 
# AgentOfferClient is @runtime_checkable
assert isinstance(client, AgentOfferClient)  # Works at runtime

Event Tracking

report_click

Report a click event when the user interacts with a tracking link:

from agentoffernetwork import ClickEvent
 
# In live mode, use the exact AON tracking URL supplied by your integration layer.
# Do not replace it with the advertiser destination (`offer.action.payload.target`).
click_result = await client.report_click(ClickEvent(
    offer_id=offer.offer_id,
    tracking_url="https://tracking.example.test/click/mock-track-1",
    timestamp="2026-04-01T12:00:00Z",
    agent_id="my-agent-v1",
    session_id="session-123",
    context={"conversation_id": "conv-456"},
))
 
print(f"Tracking ID: {click_result.tracking_id}")
print(f"Timestamp: {click_result.timestamp}")

When to report events

Click: When the user clicks the tracking URL or you programmatically redirect them. Conversion: Conversions are not reported by the SDK. The Partner that fulfils the offer reports them server-side through POST /v1/postback/{partner_id} (see Postbacks & Attribution); the tracking_id from report_click is what ties that postback back to this click.

format_recommendation

Format an offer as human-readable recommendation text. In the Python SDK, this is available both as a client method (client.format_recommendation()) and as a module-level function (format_recommendation()). The module-level function is recommended for convenience:

from agentoffernetwork import format_recommendation, FormatOptions
 
# Default: markdown style
markdown = format_recommendation(offer)
 
# Brief: single-line compact
brief = format_recommendation(offer, FormatOptions(style="brief"))
 
# Detailed: multi-line with full info
detailed = format_recommendation(offer, FormatOptions(style="detailed"))
 
# Custom options
custom = format_recommendation(offer, FormatOptions(
    style="markdown",
    include_disclosure=True,
    disclosure_text="Ad",
    include_price=True,
))

FormatOptions

Parameter

Description

style

FormatStyleOptional

Output style: 'brief' | 'detailed' | 'markdown'. Default: 'markdown'.

include_disclosure

boolOptional

Whether to include a disclosure identifier. Default: True.

disclosure_text

strOptional

Custom disclosure text. Default: 'Sponsored'.

include_price

boolOptional

Whether to include price information. Default: True.

Rendering Multiple Offers

response = await client.query_offers(params)
 
recommendations = []
for i, offer in enumerate(response.offers, 1):
    formatted = format_recommendation(offer, FormatOptions(style="markdown"))
    recommendations.append(f"### {i}.\n{formatted}")
 
print("\n\n".join(recommendations))

Error Handling

All SDK errors extend the AonError base class:

from agentoffernetwork import (
    AonError, AonAuthError, AonRateLimitError,
    AonNetworkError, AonValidationError, AonApiError,
)
 
try:
    response = await client.query_offers(params)
except AonAuthError as e:
    # Invalid or expired API key
    print(f"Auth error: {e}")
except AonRateLimitError as e:
    # Too many requests -- back off and retry
    print(f"Rate limited: {e}")
except AonNetworkError as e:
    # Timeout, DNS failure, etc.
    print(f"Network error: {e}")
except AonValidationError as e:
    # Invalid parameters
    print(f"Validation error: {e}")
except AonApiError as e:
    # Server returned a non-SUCCESS business code
    print(f"API error [{e.code}]: {e}")
except AonError as e:
    # Catch-all for any AON error
    print(f"AON error [{e.code}]: {e}")

Error Class Hierarchy

Class	Code	When
`AonError`	varies	Base class for all SDK errors
`AonAuthError`	`AUTH_ERROR`	Invalid or expired API key
`AonRateLimitError`	`RATE_LIMIT`	Request rate limit exceeded
`AonNetworkError`	`NETWORK_ERROR`	Timeout, DNS failure, connection error
`AonValidationError`	`VALIDATION_ERROR`	Invalid parameters passed to SDK methods
`AonApiError`	varies	Server returned a non-SUCCESS business code

Retry with Exponential Backoff

import asyncio
from agentoffernetwork import AonRateLimitError, AonNetworkError
 
async def query_with_retry(client, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await client.query_offers(params)
        except (AonRateLimitError, AonNetworkError):
            if attempt < max_retries - 1:
                delay = (2 ** attempt) * 1.0
                await asyncio.sleep(delay)
                continue
            raise
    raise RuntimeError("Max retries exceeded")

Mock Mode

Mock mode returns realistic fake data without making real API calls:

Development with Mock Mode

from agentoffernetwork import initialize, SDKConfig
 
client = await initialize(SDKConfig(
    api_key="any-key-works-in-mock",
    mode="mock",
))
 
# Works exactly like live mode
response = await client.query_offers(params)

Testing with MockAgentOfferClient

In mock mode, the client is actually a MockAgentOfferClient with additional test helper methods:

from agentoffernetwork import initialize, SDKConfig, MockAgentOfferClient, ClickEvent
 
client = await initialize(SDKConfig(api_key="test-key", mode="mock"))
 
# Type assertion for test helpers
assert isinstance(client, MockAgentOfferClient)
 
# Perform operations
response = await client.query_offers(params)
await client.report_click(ClickEvent(
    offer_id=response.offers[0].offer_id,
    tracking_url="https://tracking.example.test/click/mock-track-1",
    timestamp="2026-04-01T12:00:00Z",
))
 
# Verify recorded events
events = client.get_recorded_events()
assert len(events) == 1
assert events[0].type == "click"
print(events[0].timestamp)

Mock mode tip

get_recorded_events() returns all click and conversion events recorded during the session. Use this in unit tests to verify your integration logic without hitting real APIs.

API Differences from TypeScript SDK

Aspect	TypeScript	Python
Naming convention	camelCase (`queryOffers`)	snake_case (`query_offers`)
Async pattern	`async/await` (Promise)	`async/await` (coroutine)
Type system	Interfaces + type aliases	`@dataclass` + `Literal` types
Client protocol	`interface AgentOfferClient`	`@runtime_checkable Protocol`
formatRecommendation	`client.formatRecommendation()` (instance method)	`client.format_recommendation()` (instance method) or `format_recommendation()` (module-level function)
Config type	`SDKConfig` interface	`SDKConfig` dataclass
Null handling	`null` type union	`None` with `

API Reference

For the complete API surface, see the Python SDK Reference.

Next Steps

TypeScript SDK Guide -- Same concepts, TypeScript-specific patterns
Skill/Plugin Development -- MCP-based integration
Best Practices -- Production patterns, security, and FAQ

Start from Quick Start