Multi-platform social media posting service with automatic provider failover. Handles posting to 9 platforms (Twitter/X, LinkedIn, Instagram, Facebook, TikTo...
---
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.
homepage: https://canlah.ai
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
```python
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:
```python
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
```python
POST /social-accounts/auth-url
{
"platform": "twitter",
"redirect_url": "https://your-app.com/callback"
}
# Returns: {"url": "..."} or {"data": {"auth_url": "..."}}
```
### Get Connected Accounts
```python
GET /social-accounts
# Returns: {"data": [{"id": "...", "platform": "twitter", "username": "..."}]}
```
### Media Upload (Presigned URL Flow)
```python
# 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
```python
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
```python
GET /accounts
# Returns: {"accounts": [{"_id": "...", "platform": "...", "username": "...", "profileId": "..."}]}
```
### Media Upload (Presigned URL Flow)
```python
# 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
```python
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
```bash
# 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
```python
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
```python
# 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
```python
# 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
```python
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
```python
# 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
```python
# 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
```sql
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
```sql
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)
```python
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
---
## Author
**[Canlah AI](https://canlah.ai)** — Run performance marketing without breaking your brand.
- GitHub: [github.com/PHY041](https://github.com/PHY041)
- All Skills: [clawhub.ai/PHY041](https://clawhub.ai/PHY041)
don't have the plugin yet? install it then click "run inline in claude" again.
Postiz is a tool to schedule social media and chat posts to 28+ channels X, LinkedIn, LinkedIn Page, Reddit, Instagram, Facebook Page, Threads, YouTube, Goog...