🧪 Skills

Social Posting

Multi-platform social media posting service with automatic provider failover. Handles posting to 9 platforms (Twitter/X, LinkedIn, Instagram, Facebook, TikTo...

v1.0.0
❤️ 0
⬇️ 94
👁 1
Share

Description


name: social-posting description: Multi-platform social media posting service with automatic provider failover. Handles posting to 9 platforms (Twitter/X, LinkedIn, Instagram, Facebook, TikTok, Threads, Bluesky, YouTube, Pinterest) with per-user credential management, OAuth flow, media upload, scheduling, and post history tracking. Triggers on "post to social media", "publish to platforms", "schedule a post", "social post", or any cross-platform publishing request. metadata: {"category": "social-media", "platforms": ["twitter", "linkedin", "instagram", "facebook", "tiktok", "threads", "bluesky", "youtube", "pinterest"], "providers": ["PostForMe", "LATE"], "features": ["oauth", "scheduling", "media-upload", "failover", "post-history"]}

Social Posting Skill

Multi-platform social media posting with automatic provider failover. Users connect their own social accounts via OAuth — credentials are encrypted at rest. Supports immediate posting, scheduling, and media attachments.


Supported Platforms

Platform Enum Value
Twitter / X twitter
LinkedIn linkedin
Instagram instagram
Facebook facebook
TikTok tiktok
Threads threads
Bluesky bluesky
YouTube youtube
Pinterest pinterest

Provider Architecture

Two providers with automatic failover:

Provider Role API Base
PostForMe Primary (cheaper) https://api.postforme.dev/v1
LATE Fallback (reliable) https://getlate.dev/api/v1

Failover order:

  1. Try user's PostForMe credentials
  2. Try user's LATE credentials
  3. Fall back to global env-var credentials (PostForMe → LATE)

Core Data Structures

class Platform(Enum):
    TWITTER = "twitter"
    LINKEDIN = "linkedin"
    INSTAGRAM = "instagram"
    FACEBOOK = "facebook"
    TIKTOK = "tiktok"
    THREADS = "threads"
    BLUESKY = "bluesky"
    YOUTUBE = "youtube"
    PINTEREST = "pinterest"

@dataclass
class PostResult:
    success: bool
    post_id: Optional[str] = None
    platform_post_ids: Optional[Dict[str, str]] = None
    platform_post_urls: Optional[Dict[str, str]] = None
    error: Optional[str] = None
    provider: Optional[str] = None
    scheduled_for: Optional[datetime] = None

@dataclass
class AccountInfo:
    id: str
    platform: str
    username: str
    profile_id: Optional[str] = None

Provider Interface

Both providers implement the same abstract interface:

class SocialPostingProvider(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...

    @abstractmethod
    def get_accounts(self) -> List[AccountInfo]: ...

    @abstractmethod
    def upload_media(self, image_url: str) -> Optional[str]: ...

    @abstractmethod
    def post(
        self,
        content: str,
        platforms: List[str],
        media_urls: Optional[List[str]] = None,
        scheduled_for: Optional[datetime] = None
    ) -> PostResult: ...

PostForMe Provider (Primary)

Authentication: Authorization: Bearer {api_key}

OAuth URL Generation

POST /social-accounts/auth-url
{
  "platform": "twitter",
  "redirect_url": "https://your-app.com/callback"
}
# Returns: {"url": "..."} or {"data": {"auth_url": "..."}}

Get Connected Accounts

GET /social-accounts
# Returns: {"data": [{"id": "...", "platform": "twitter", "username": "..."}]}

Media Upload (Presigned URL Flow)

# 1. Get presigned URL
POST /media/create-upload-url
{"content_type": "image/jpeg"}
# Returns: {"upload_url": "...", "media_url": "..."}

# 2. PUT image bytes to upload_url
# 3. Use returned media_url in post payload

Create Post

POST /social-posts
{
  "caption": "Post content here",
  "social_accounts": ["account_id_1", "account_id_2"],
  "media": [{"url": "https://..."}],        # optional
  "scheduled_at": "2025-01-01T09:00:00"     # optional ISO datetime
}
# Returns: {"id": "post_id"} or {"data": {"id": "post_id"}}

Note: PostForMe normalizes twitter platform to both "twitter" and "x" internally.


LATE Provider (Fallback)

Authentication: Authorization: Bearer {api_key}

Get Connected Accounts

GET /accounts
# Returns: {"accounts": [{"_id": "...", "platform": "...", "username": "...", "profileId": "..."}]}

Media Upload (Presigned URL Flow)

# 1. Get presigned URL
POST /media/presign
{"filename": "media.jpg", "contentType": "image/jpeg"}
# Returns: {"uploadUrl": "...", "publicUrl": "..."}

# 2. PUT image bytes to uploadUrl (wait 1s for CDN propagation)
# 3. Use publicUrl in post payload

Create Post

POST /posts
{
  "content": "Post content here",
  "platforms": [
    {"platform": "twitter", "accountId": "...", "profileId": "..."}
  ],
  "mediaItems": [{"url": "https://...", "type": "image"}],  # optional
  "scheduledFor": "2025-01-01T09:00:00"                      # optional
}
# Returns: {"post": {"_id": "...", "platforms": [{"platform": "twitter", "platformPostId": "...", "platformPostUrl": "..."}]}}

Service Layer (SocialPostingService)

The service layer wraps both providers and adds database integration.

Environment Variables Required

# Provider API keys (for global fallback if user has no personal creds)
POSTFORME_API_KEY=your_postforme_key
LATE_API_KEY=your_late_key

# Encryption for stored credentials
ENCRYPTION_KEY=your_fernet_key  # Generate: Fernet.generate_key()

Credential Management

service = SocialPostingService()
service.init(supabase_client)

# Save user credentials
service.save_credentials(
    user_id="user-uuid",
    provider="postforme",  # or "late"
    api_key="sk-...",
    connected_platforms=["twitter", "linkedin"]
)

# Get credentials (auto-decrypted)
creds = service.get_credentials(user_id="user-uuid", provider="postforme")

# Delete credentials
service.delete_credentials(user_id="user-uuid", provider="postforme")

Credentials are encrypted using Fernet symmetric encryption before database storage. Set ENCRYPTION_KEY environment variable to a valid Fernet key.

OAuth Flow

# Generate OAuth URL for user to connect a platform
oauth_url = service.get_oauth_url(
    user_id="user-uuid",
    platform="twitter",
    redirect_url="https://your-app.com/oauth/callback"
)
# Returns: URL string or None

Posting

# Immediate post
result = service.create_post(
    user_id="user-uuid",
    content="Your post content",
    platforms=["twitter", "linkedin"],
    media_urls=["https://cdn.example.com/image.jpg"],  # optional
    scheduled_for=None,
    campaign_id="campaign-uuid",   # optional, for tracking
    batch_number=1                 # optional, for tracking
)

# Scheduled post
from datetime import datetime, timezone
result = service.create_post(
    user_id="user-uuid",
    content="Scheduled post content",
    platforms=["linkedin"],
    scheduled_for=datetime(2025, 6, 1, 9, 0, 0, tzinfo=timezone.utc)
)

# result.success → bool
# result.post_id → provider post ID
# result.platform_post_ids → {"twitter": "tweet_id", ...}
# result.platform_post_urls → {"twitter": "https://...", ...}
# result.error → error message if failed
# result.provider → "PostForMe" or "LATE"

Publish from Campaign Calendar

result = service.publish_batch(
    user_id="user-uuid",
    campaign_id="campaign-uuid",
    batch_number=3,
    platforms=["twitter", "instagram"],
    media_urls=["https://..."],    # selected images for this batch
    scheduled_for=None             # or datetime for scheduling
)

Looks up campaigns.creative_calendar.batches[n].caption from database and posts it.

Account Management

# Get connected platforms for a user
accounts = service.get_connected_accounts(user_id="user-uuid")
# Returns: [{"id": "...", "platform": "twitter", "username": "@handle"}]

# Refresh connected_platforms field in credentials table
service.refresh_connected_platforms(user_id="user-uuid")

Post History

# Get all posts
history = service.get_post_history(user_id="user-uuid", limit=50)

# Filter by status: "posted", "scheduled", "failed"
scheduled = service.get_post_history(user_id="user-uuid", status="scheduled")

# Get single post
post = service.get_post(post_id="post-uuid")

Database Schema

user_social_credentials

CREATE TABLE user_social_credentials (
    user_id         UUID NOT NULL,
    provider        TEXT NOT NULL,  -- 'postforme' | 'late'
    encrypted_api_key TEXT NOT NULL,
    connected_platforms TEXT[],
    updated_at      TIMESTAMPTZ,
    PRIMARY KEY (user_id, provider)
);

social_posts

CREATE TABLE social_posts (
    id                  UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    user_id             UUID NOT NULL,
    provider            TEXT,
    provider_post_id    TEXT,
    platforms           TEXT[],
    platform_post_ids   JSONB,
    platform_post_urls  JSONB,
    content             TEXT,
    media_urls          TEXT[],
    scheduled_for       TIMESTAMPTZ,
    posted_at           TIMESTAMPTZ,
    status              TEXT,  -- 'posted' | 'scheduled' | 'failed'
    error_message       TEXT,
    campaign_id         UUID,
    batch_number        INTEGER,
    created_at          TIMESTAMPTZ DEFAULT NOW()
);

Failover Logic

User requests post to platforms X, Y, Z
         ↓
Check: user has PostForMe creds?
    YES → PostForMeProvider(user_key)
    NO  → Check: user has LATE creds?
              YES → LateProvider(user_key)
              NO  → Check: POSTFORME_API_KEY env var?
                        YES → PostForMeProvider(global_key)
                        NO  → Check: LATE_API_KEY env var?
                                  YES → LateProvider(global_key)
                                  NO  → Return error: no credentials configured
         ↓
Call provider.post(content, platforms, media_urls, scheduled_for)
         ↓
Track result in social_posts table
         ↓
Return PostResult

Usage Example (Standalone)

import os
from datetime import datetime
from social_posting_service import (
    SocialPostingService,
    PostForMeProvider,
    LateProvider
)

# Option A: Use PostForMe directly
provider = PostForMeProvider(api_key=os.getenv("POSTFORME_API_KEY"))
result = provider.post(
    content="Hello from the API!",
    platforms=["twitter", "linkedin"],
    media_urls=["https://example.com/image.jpg"]
)
print(result.success, result.post_id)

# Option B: Use service with database
from supabase import create_client
supabase = create_client(os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_KEY"))

service = SocialPostingService()
service.init(supabase)
result = service.create_post(
    user_id="user-uuid",
    content="Post via service",
    platforms=["twitter"]
)

Error Handling

All methods return structured results — they do not raise exceptions to the caller.

Error Condition Result
No credentials configured PostResult(success=False, error="No social posting credentials configured...")
No connected account for platform PostResult(success=False, error="No connected accounts for: [...]")
Provider HTTP error PostResult(success=False, error="HTTP 4xx: ...")
Network timeout PostResult(success=False, error="...")
DB tracking failure Logs error, still returns posting result

Integration Checklist

  • Install requests, cryptography packages
  • Set ENCRYPTION_KEY env var (generate with Fernet.generate_key())
  • Set at least one provider key: POSTFORME_API_KEY or LATE_API_KEY
  • Create user_social_credentials table in database
  • Create social_posts table in database
  • Initialize service: service.init(supabase_client)
  • Guide users through OAuth flow before first post

Reviews (0)

Sign in to write a review.

No reviews yet. Be the first to review!

Comments (0)

Sign in to join the discussion.

No comments yet. Be the first to share your thoughts!

Compatible Platforms

Pricing

Free

Related Configs