Social media API from Python

pip install aether

# Set your API key in the environment
export AETHER_API_KEY="your_api_key_here"

# Or use a .env file with python-dotenv:
# pip install python-dotenv
# AETHER_API_KEY=your_api_key_here

import os
from aether import Aether

client = Aether(api_key=os.environ["AETHER_API_KEY"])

# Verify the connection
profiles = client.profiles.list()
print(f"Connected profiles: {len(profiles.data)}")

import os
import webbrowser
from aether import Aether

client = Aether(api_key=os.environ["AETHER_API_KEY"])

# Generate an OAuth Connect Link for Instagram
link = client.connect_links.create(
    platform="instagram",
    # After authentication, Instagram redirects here
    redirect_url="https://your-app.com/oauth/callback",
)

print(f"Connect URL: {link.url}")
# Open in the browser so the user can authenticate
webbrowser.open(link.url)

# After OAuth, the profile appears immediately:
profiles = client.profiles.list()
instagram = next(p for p in profiles.data if p.platform == "instagram")
print(f"Connected: {instagram.display_name} ({instagram.id})")
# ig_abc123

import os
from aether import Aether

client = Aether(api_key=os.environ["AETHER_API_KEY"])

# Post immediately to multiple platforms
post = client.posts.create(
    text="New release is live — check the changelog 🚀",
    profile_ids=[
        "ig_abc123",   # Instagram
        "li_comp789",  # LinkedIn
        "th_user456",  # Threads
    ],
    media_urls=["https://your-cdn.com/release-screenshot.png"],
)

print(post.data.id)      # post_abc123
print(post.data.status)  # "publishing"

import os
from datetime import datetime, timezone, timedelta
from aether import Aether

client = Aether(api_key=os.environ["AETHER_API_KEY"])

# Schedule for tomorrow at 9am UTC
tomorrow_9am = (
    datetime.now(timezone.utc).replace(hour=9, minute=0, second=0, microsecond=0)
    + timedelta(days=1)
).isoformat()

post = client.posts.create(
    text="Good morning — here's what we shipped this week.",
    profile_ids=["ig_abc123", "li_comp789"],
    scheduled_for=tomorrow_9am,  # "2026-07-16T09:00:00+00:00"
)

print(post.data.status)       # "scheduled"
print(post.data.scheduled_for)  # 2026-07-16T09:00:00+00:00

# Per-platform overrides — tailor content per profile in one call
post = client.posts.create(
    text="Default caption — shown if no override is set for a profile.",
    profile_ids=["ig_abc123", "li_comp789", "tt_xyz789"],
    overrides={
        "ig_abc123": {
            "text": "Launch day 🎉 #buildinpublic #saas #startup",
        },
        "li_comp789": {
            "text": (
                "Today we shipped a major product update. "
                "Here's what changed and what's coming next:"
            ),
        },
        "tt_xyz789": {
            "text": "we shipped something wild. let me explain 👀",
        },
    },
)

import os
from aether import Aether

client = Aether(api_key=os.environ["AETHER_API_KEY"])

# Account-level metrics for the last 30 days
metrics = client.analytics.get_account_metrics(
    from_date="2026-06-01",
    to_date="2026-06-30",
    granularity="day",
    platform="instagram",  # optional — omit to list all connected profiles
)

for point in metrics.data:
    print(f"{point.date}: {point.impressions} impressions, {point.engagement} engagements")

# Post-level metrics
post_metrics = client.analytics.get_post_metrics("post_abc123")
for m in post_metrics.data:
    print(f"{m.platform}: {m.metric_type} = {m.value}")

# Read and reply to DMs for connected accounts
messages = client.inbox.list(status="unread", limit=20)

for msg in messages.data:
    print(f"[{msg.platform}] {msg.sender_name}: {msg.text}")

    # Reply to the message
    if "pricing" in msg.text.lower():
        client.inbox.reply(
            message_id=msg.id,
            text="Thanks for reaching out! Here's our pricing: https://aetherhq.dev/pricing",
        )

# Flask webhook handler with HMAC verification
import os
import hmac
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)

def verify_signature(body: bytes, signature: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.route("/webhooks/aether", methods=["POST"])
def webhook():
    sig = request.headers.get("X-Aether-Signature", "")

    if not verify_signature(request.data, sig, os.environ["AETHER_WEBHOOK_SECRET"]):
        return jsonify({"error": "Invalid signature"}), 401

    payload = request.get_json(force=True)
    event   = payload.get("event")
    data    = payload.get("data", {})

    if event == "post.published":
        print(f"Published on {data['platform']}: {data.get('platformPostUrl')}")
        # Update your DB, notify users, etc.

    elif event == "post.failed":
        print(f"Failed on {data['platform']}: {data['error']['message']}")
        # Alert the user, schedule retry, etc.

    elif event == "message.created":
        print(f"New DM from {data['senderName']} on {data['platform']}")
        # Trigger auto-reply flow, alert support team, etc.

    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(port=5000)

# Async usage with httpx (Aether's Python SDK supports async)
import os
import asyncio
from aether import AsyncAether

async def main():
    client = AsyncAether(api_key=os.environ["AETHER_API_KEY"])

    # Post to 3 platforms concurrently
    post = await client.posts.create(
        text="Shipped something new today 🚀",
        profile_ids=["ig_abc123", "li_comp789", "th_user456"],
        scheduled_for="2026-07-20T09:00:00Z",
    )
    print(post.data.status)

    # Pull analytics while waiting on post confirmation
    metrics, profiles = await asyncio.gather(
        client.analytics.get_account_metrics(
            from_date="2026-07-01",
            to_date="2026-07-14",
        ),
        client.profiles.list(),
    )
    print(f"{len(profiles.data)} profiles, {len(metrics.data)} data points")

asyncio.run(main())