ads4gpts-core 0.1.1

Python SDK for integrating Ads4GPTs advertising logic into conversational AI applications via LangGraph agents

Ads4GPTs Core SDK

Python SDK for integrating with Ads4GPTs, the monetization backbone of the AI Native Internet.

Overview

The Ads4GPTs Core SDK provides the tools to configure your Integration Agent, connecting your AI application to the Ads4GPTs platform. Ad requests are initiated autonomously based on conversational context and user intent, with full support for streaming responses.

Installation

pip install ads4gpts-core

Requirements

• Python 3.11 or higher
• Active Ads4GPTs account with API credentials

Quick Start

1. Configure Environment Variables

export ADS4GPTS_AGENT_URL="https://your-agent-api.com"
export ADS4GPTS_GPT_ID="your-gpt-id"
export ADS4GPTS_API_KEY="your-api-key"
export ADS4GPTS_SSP_ID="your-ssp-id"

2. Basic Usage

from ads4gpts_core import invoke_integration_agent

# First conversation in a session
stream = invoke_integration_agent(
    user_input="What are the best noise-cancelling headphones under €200?",
    message_history=[],
    thread_id="thread_123",
    session_id="session_456",
)

# Process streaming response
async for chunk in stream:
    print(chunk)

3. Continuing a Conversation

# Subsequent messages in the same thread
stream = invoke_integration_agent(
    user_input="How about wireless options?",
    message_history=[
        {"role": "human", "content": "What are the best noise-cancelling headphones under €200?"},
        {"role": "ai", "content": "Here are some great options..."}
    ],
    thread_id="thread_123",
    session_id="session_456",
)

async for chunk in stream:
    print(chunk)

Stream Processing Utilities

Split Stream

Duplicate an async stream into two independent consumers:

from ads4gpts_core import split_stream

async def source_stream():
    for i in range(5):
        yield i

streams = split_stream(source_stream())

# Both consumers receive all chunks
async for chunk in streams['app_stream']:
    print(f"App: {chunk}")

async for chunk in streams['util_stream']:
    print(f"Util: {chunk}")

Merge Streams

Process two streams sequentially:

from ads4gpts_core import merge_streams

async def app_stream():
    for i in range(3):
        yield f"app_{i}"

async def ads_stream():
    for i in range(3):
        yield f"ads_{i}"

# Outputs: app_0, app_1, app_2, ads_0, ads_1, ads_2
async for chunk in merge_streams(app_stream(), ads_stream()):
    print(chunk)