ADS Publisher

The ADS Publisher allows you to send events to the ADS Subscribers for AI agents to react. This guide covers the essential methods and configuration options for publishing events using the Python SDK.

Getting Started

Before you start publishing events, ensure you have the ADS Bridge set up and running.

Installation


pip install agentdatashuttle-adspy

Initialize Your Publisher

First, set up the connection parameters and create an instance of the ADS Publisher:


import os
from agentdatashuttle_adspy.core.publisher import ADSPublisher
from agentdatashuttle_adspy.core.client import ADSClientParams
from agentdatashuttle_adspy.models.models import ADSDataPayload
 
# Configure RabbitMQ connection parameters that the ADS Bridge also connects to
client_params = ADSClientParams(
    host=os.getenv("RABBITMQ_HOST", "localhost"),
    username=os.getenv("RABBITMQ_USERNAME", "ads_user"),
    password=os.getenv("RABBITMQ_PASSWORD", "ads_password"),
    port=int(os.getenv("RABBITMQ_PORT", 5672))
)
 
# Create publisher instance
publisher = ADSPublisher("YourEventSource", client_params)

The eventSource parameter could be any identifier for your event data source, such as “NotionEvents” or “StripeEvents”. It just serves as a reference.

Publish Your Event

Create an event payload and publish it to the ADS Bridge:


# Create event payload
payload = ADSDataPayload(
    event_name="system_alert",
    event_description="Critical system event occurred",
    event_data={
        "timestamp": "2025-07-09T10:30:00Z",
        "severity": "high"
    }
)
 
# Publish the event
publisher.publish_event(payload)
print("Event published successfully.")

The event_data field could be any structured data relevant to your event, such as user actions, system metrics, business transactions, etc. to match your use case.

💡

Add a straightforward event_name and a detailed event_description to improve the results of the AI agents at the subscriber end (To learn more about this refer here). This also helps provide better reports for the subscribers after the agent invocation is complete.

Configuration

Client Parameters

The ADSClientParams class defines the connection settings for your RabbitMQ server that the Publisher and Bridge connect to:


from agentdatashuttle_adspy.core.client import ADSClientParams
 
client_params = ADSClientParams(
    host="your-rabbitmq-host",        # RabbitMQ server hostname
    username="your-rabbitmq-username", # RabbitMQ username
    password="your-rabbitmq-password", # RabbitMQ password
    port=5672                         # RabbitMQ port
)

Use environment variables to store sensitive connection details like passwords and hostnames.

ADSPublisher

Constructor


ADSPublisher(event_source: str, client_params: ADSClientParams)

Parameters:

event_source (str): Identifier for your event source (e.g., “UptimeKumaEvents”, “StripeEvents”)
client_params (ADSClientParams): Connection configuration object

Methods

publish_event()

Publishes an event to the ADS Bridge for processing.


publisher.publish_event(payload: ADSDataPayload) -> None

Parameters:

payload (ADSDataPayload): The event data to publish

Data Payload Structure

ADSDataPayload Class


from agentdatashuttle_adspy.models.models import ADSDataPayload
 
payload = ADSDataPayload(
    event_name="string",        # Short identifier for the event type
    event_description="string", # Human-readable event description
    event_data={}              # Custom event data dictionary
)

Event Data Examples

System Monitoring


payload = ADSDataPayload(
    event_name="cpu_threshold_exceeded",
    event_description="Server CPU usage exceeded 80% threshold on server 'srv-001'",
    event_data={
        "server_id": "srv-001",
        "cpu_usage": 85.2,
        "memory_usage": "2042Mi",
        "timestamp": "2025-07-09T10:30:00Z",
        "alert_level": "warning"
    }
)

Container Management


payload = ADSDataPayload(
    event_name="container_down",
    event_description="The argocd service is down. Container 'argocd-server' in namespace 'argocd-system' has failed with last memory captured as '2042Mi'.",
    event_data={
        "service_name": "argocd",
        "timestamp": "23-04-2004",
        "memory_captured_last": "2042Mi",
        "namespace": "argocd-system",
        "pod_status": "Failed"
    }
)

Business Events


payload = ADSDataPayload(
    event_name="subscription_upgraded",
    event_description="Customer with customer_id 'cust_789' upgraded their subscription plan from 'basic' to 'premium'. This will increase the monthly revenue by $25.00.",
    event_data={
        "customer_id": "cust_789",
        "old_plan": "basic",
        "new_plan": "premium",
        "upgrade_date": "2025-07-09T10:30:00Z",
        "monthly_revenue_impact": 25.00
    }
)

Error Handling

⚠️

Always implement proper error handling when publishing events to ensure your application remains stable.


try:
    publisher.publish_event(payload)
    print("Event published successfully")
except Exception as error:
    print(f"Publishing failed: {error}")
    # Implement retry logic or fallback mechanism

Logging

Configure logging level via the LOG_LEVEL environment variable:

Level	Description
error	Critical errors that may cause the app to crash
warn	Warnings about potentially harmful situations
info	General operational information
debug	Debug-level logs for development

Next Steps

Explore the SDK usage for the ADS Subscriber to consume events
Explore Node.js SDK for Node.js applications
Learn about n8n Integration for workflow automation