Social Posting
Multi-platform social media posting service with automatic provider failover. Handles posting to 9 platforms (Twitter/X, LinkedIn, Instagram, Facebook, TikTo...
Like a lobster shell, security has layers — review code before you run it.
License
SKILL.md
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 | |
instagram | |
facebook | |
| TikTok | tiktok |
| Threads | threads |
| Bluesky | bluesky |
| YouTube | youtube |
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:
- Try user's PostForMe credentials
- Try user's LATE credentials
- 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,cryptographypackages - Set
ENCRYPTION_KEYenv var (generate withFernet.generate_key()) - Set at least one provider key:
POSTFORME_API_KEYorLATE_API_KEY - Create
user_social_credentialstable in database - Create
social_poststable in database - Initialize service:
service.init(supabase_client) - Guide users through OAuth flow before first post
Files
1 totalComments
Loading comments…
