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.
post content to one or more social platforms simultaneously with automatic provider failover. use this when you need to publish a message, image, or scheduled post across twitter, linkedin, instagram, facebook, tiktok, threads, bluesky, youtube, or pinterest. the skill manages user oauth credentials (encrypted), handles media uploads, schedules posts for later, and tracks post history. if the primary provider fails, it automatically retries with the fallback provider.
user context
user_id: uuid of the user posting. required.content: text of the post (max length depends on platform, typically 280-5000 chars). required.platforms: list of platform enum values (e.g., ["twitter", "linkedin"]). at least one required.media_urls: optional list of publicly accessible image/video urls to attach. platform-specific limits apply (twitter: 4 images, tiktok/youtube: 1 video, etc.).scheduled_for: optional iso 8601 datetime (e.g., 2025-06-01T09:00:00Z) to schedule post for later. if null, posts immediately.campaign_id: optional uuid for tracking posts to campaigns. stored in social_posts.campaign_id.batch_number: optional integer for batch publishing from campaign calendars. used with publish_batch().external connections
https://api.postforme.dev/v1. requires POSTFORME_API_KEY env var or per-user oauth token.https://getlate.dev/api/v1. requires LATE_API_KEY env var or per-user oauth token.user_social_credentials, post history in social_posts. requires SUPABASE_URL and SUPABASE_KEY env vars.environment variables
POSTFORME_API_KEY=sk-... # global api key for PostForMe (fallback if user has no personal creds)
LATE_API_KEY=... # global api key for LATE (fallback if user has no personal creds)
ENCRYPTION_KEY=... # fernet symmetric key for encrypting stored api keys. generate with Fernet.generate_key()
SUPABASE_URL=https://... # postgres-compatible database
SUPABASE_KEY=eyJ0... # supabase anon/service key
database tables
user_social_credentials (user_id, provider, encrypted_api_key, connected_platforms, updated_at). stores encrypted oauth/api tokens.social_posts (id, user_id, provider, provider_post_id, platforms, platform_post_ids, platform_post_urls, content, media_urls, scheduled_for, posted_at, status, error_message, campaign_id, batch_number, created_at). tracks all posting attempts.validate inputs. check that user_id is provided, content is not empty, platforms list is not empty. if any fail, return PostResult(success=False, error="...").
get user credentials. call get_credentials(user_id, "postforme"). if found and not expired, extract api_key. if not found, set to null. repeat for "late" provider. if neither user credential exists, use global POSTFORME_API_KEY and LATE_API_KEY from env vars.
select primary provider and api_key. use failover order: user's postforme creds > user's late creds > global postforme key > global late key. if no credentials available at all, return error.
upload media (if media_urls provided). for each media_url in media_urls: call provider's upload_media(media_url) to get a provider-hosted media_id or url. store results. if media upload fails for any url, return PostResult(success=False, error="Media upload failed: ...") and stop.
call provider.post(). invoke the selected provider's post method with: content, platforms (list), media (list of uploaded media ids/urls), scheduled_for (datetime or null). provider returns a PostResult with success flag, post_id, platform_post_ids dict, platform_post_urls dict, and any error message.
on provider success. write result to social_posts table: insert row with user_id, provider name, provider_post_id, platforms array, platform_post_ids (jsonb), platform_post_urls (jsonb), content, media_urls, scheduled_for, status ("posted" or "scheduled"), campaign_id, batch_number, created_at. return result to caller.
on provider failure (if primary provider used). if result.success is false and primary provider was postforme, retry with late provider using same inputs. if retry succeeds, write to db and return success result. if retry also fails, write both attempt failures to db (log the second error) and return the final failure result.
on provider failure (if no fallback available). log error, write failure record to db with error_message, return PostResult(success=False, error=...).
PostResult(success=False, error="Request timeout") and attempt failover to next provider if available.success case
PostResult(
success=True,
post_id="provider-post-id",
platform_post_ids={"twitter": "12345", "linkedin": "67890"},
platform_post_urls={"twitter": "https://twitter.com/user/status/12345", "linkedin": "https://linkedin.com/feed/update/..."},
error=None,
provider="PostForMe", # or "LATE"
scheduled_for=None # or datetime if scheduled
)
failure case
PostResult(
success=False,
post_id=None,
platform_post_ids=None,
platform_post_urls=None,
error="HTTP 401: Unauthorized" or "No credentials configured" or "No connected accounts for: [instagram, tiktok]",
provider=None,
scheduled_for=None
)
database record (one row inserted into social_posts per posting attempt)
id: uuid
user_id: uuid
provider: text ("PostForMe" or "LATE" or null if no provider attempted)
provider_post_id: text (null if failed)
platforms: text[] (e.g., ["twitter", "linkedin"])
platform_post_ids: jsonb (e.g., {"twitter": "12345", "linkedin": "67890"} or null)
platform_post_urls: jsonb (e.g., {"twitter": "https://...", "linkedin": "https://..."} or null)
content: text
media_urls: text[] (original urls from input)
scheduled_for: timestamptz (null if posted immediately)
posted_at: timestamptz (null if scheduled or failed)
status: text ("posted", "scheduled", or "failed")
error_message: text (null if success)
campaign_id: uuid (null if not provided)
batch_number: int (null if not provided)
created_at: timestamptz (auto-set)
success: true. posted immediately or scheduled successfully.success: false and an error message explaining why (no credentials, no connected account, network error, etc.). no post was sent.| platform | enum value | notes |
|---|---|---|
| twitter / x | twitter |
postforme normalizes to both "twitter" and "x" internally |
linkedin |
requires connected linkedin account | |
instagram |
platform limits: 1 image or carousel (up to 10 images) | |
facebook |
posts to user's feed or connected pages | |
| tiktok | tiktok |
requires video media (not text-only) |
| threads | threads |
text and image support |
| bluesky | bluesky |
atproto protocol backend |
| youtube | youtube |
requires video media |
pinterest |
image-focused, requires media |
two providers with automatic failover and identical abstract interface:
| provider | role | api base | auth |
|---|---|---|---|
| PostForMe | primary (cheaper, faster) | https://api.postforme.dev/v1 |
bearer token in Authorization: Bearer {api_key} header |
| LATE | fallback (more reliable) | https://getlate.dev/api/v1 |
bearer token in Authorization: Bearer {api_key} header |
failover order (tried in sequence until success):
user_social_credentials)user_social_credentials)POSTFORME_API_KEY env varLATE_API_KEY env varoauth url generation
POST /social-accounts/auth-url
{
"platform": "twitter",
"redirect_url": "https://your-app.com/callback"
}
# Response: {"url": "https://..."} or {"data": {"auth_url": "https://..."}}
get connected accounts
GET /social-accounts
Headers: Authorization: Bearer {api_key}
# Response: {"data": [{"id": "acc-123", "platform": "twitter", "username": "@handle"}]}
media upload (presigned url flow)
POST /media/create-upload-url
Headers: Authorization: Bearer {api_key}
{"content_type": "image/jpeg"}
# Response: {"upload_url": "https://...", "media_url": "https://..."}
# Then: PUT raw image bytes to upload_url (no auth needed)
# Result: media_url is ready for use in post payload
create post
POST /social-posts
Headers: Authorization: Bearer {api_key}
{
"caption": "Your post content here",
"social_accounts": ["acc-id-1", "acc-id-2"],
"media": [{"url": "https://cdn.example.com/image.jpg"}], # optional
"scheduled_at": "2025-06-01T09:00:00" # optional iso datetime
}
# Response: {"id": "post-uuid"} or {"data": {"id": "post-uuid"}}
get connected accounts
GET /accounts
Headers: Authorization: Bearer {api_key}
# Response: {"accounts": [{"_id": "acc-123", "platform": "twitter", "username": "@handle", "profileId": "prof-456"}]}
media upload (presigned url flow)
POST /media/presign
Headers: Authorization: Bearer {api_key}
{"filename": "media.jpg", "contentType": "image/jpeg"}
# Response: {"uploadUrl": "https://...", "publicUrl": "https://..."}
# Then: PUT raw image bytes to uploadUrl (no auth needed)
# Wait 1 second for cdn propagation before using publicUrl in post
create post
POST /posts
Headers: Authorization: Bearer {api_key}
{
"content": "Your post content here",
"platforms": [
{"platform": "twitter", "accountId": "acc-123", "profileId": "prof-456"}
],
"mediaItems": [{"url": "https://...", "type": "image"}], # optional
"scheduledFor": "2025-06-01T09:00:00" # optional iso datetime
}
# Response: {"post": {"_id": "post-uuid", "platforms": [{"platform": "twitter", "platformPostId": "tweet-id", "platformPostUrl": "https://twitter.com/..."}]}}
save user api key (encrypted at rest)
service.save_credentials(
user_id="user-uuid",
provider="postforme", # or "late"
api_key="sk-...",
connected_platforms=["twitter", "linkedin"]
)
# encrypts api_key with ENCRYPTION_KEY, stores in user_social_credentials table
retrieve user api key (auto-decrypted)
creds = service.get_credentials(user_id="user-uuid", provider="postforme")
# Returns: {"api_key": "sk-...", "connected_platforms": ["twitter", "linkedin"], ...} or None
delete user api key
service.delete_credentials(user_id="user-uuid", provider="postforme")
# Removes row from user_social_credentials
refresh connected platforms (after user adds/removes account in provider ui)
service.refresh_connected_platforms(user_id="user-uuid")
# calls provider.get_accounts(), updates connected_platforms in credentials
immediate post to multiple platforms
from social_posting_service import SocialPostingService
from supabase import create_client
import os
supabase = create_client(os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_KEY"))
service = SocialPostingService()
service.init(supabase)
result = service.create_post(
user_id="user-123",
content="check out this new feature!",
platforms=["twitter", "linkedin"],
media_urls=["https://cdn.example.com/screenshot.png"]
)
if result.success:
print(f"posted to {len(result.platform_post_ids)} platforms")
print(f"tweet: {result.platform_post_urls.get('twitter')}")
else:
print(f"post failed: {result.error}")
schedule a post
from datetime import datetime, timezone
result = service.create_post(
user_id="user-123",
content="join us at 2pm for a live demo",
platforms=["linkedin"],
scheduled_for=datetime(2025, 6, 1, 14, 0, 0, tzinfo=timezone.utc)
)
if result.success:
print(f"scheduled for {result.scheduled_for}")
get post history
history = service.get_post_history(user_id="user-123", limit=20, status="posted")
for post in history:
print(f"{post.created_at}: {post.content[:50]}... ({post.provider})")
publish from campaign calendar (batch)
result = service.publish_batch(
user_id="user-123",
campaign_id="campaign-456",
batch_number=3,
platforms=["twitter", "instagram"],
media_urls=["https://cdn.example.com/batch3-image.jpg"]
)
# fetches caption from campaigns.creative_calendar.batches[3].caption in db
| scenario | behavior |
|---|---|
| no credentials (user or global) | return PostResult(success=False, error="No social posting credentials configured. Set POSTFORME_API_KEY or LATE_API_KEY, or save user credentials via oauth.") |
| api timeout (>30s) | catch timeout, return failure, attempt failover to next provider |
| http 401/403 (api key invalid or expired) | provider returns error, attempt failover |
| no connected account for platform | provider lookup fails, return error with platform list (e.g., "No connected accounts for: [instagram, tiktok]") |
| media upload fails | return failure before attempting post. do not retry with text-only. |
| scheduled_for is in the past | provider will reject. return error from provider. do not auto-correct. |
| network timeout on media |