Building a CrewAI crew for social media scheduling

If you’re searching for a crewai social media scheduling crew example that goes beyond a toy demo, this guide walks through a complete, runnable implementation. You’ll define three specialized agents — strategist, copywriter, and scheduler — wire them with dependent tasks, equip them with real tools for platform APIs, and verify the output at each stage. The result is a crew you can extend for multi-platform publishing, approval workflows, or content calendars.

Step 1: Project setup and dependencies

Create a fresh virtual environment and install the core packages. CrewAI sits on top of LangChain, so you’ll need both plus an LLM provider. This example uses OpenAI-compatible endpoints — swap the base URL and key for your provider of choice.

python -m venv .venv
source .venv/bin/activate
pip install crewai langchain-openai python-dotenv pydantic httpx

Create a .env file with your credentials:

OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1
# Optional: if you route through a gateway that honors cache-control hints
# OPENAI_BASE_URL=https://api.n4n.ai/v1

Verify the environment loads:

# test_env.py
import os
from dotenv import load_dotenv

load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY missing"
print("Environment OK")

Run python test_env.py — you should see “Environment OK”.

Step 2: Define the agents

Three agents cover the pipeline: a strategist who selects topics and angles, a copywriter who drafts platform-specific copy, and a scheduler that calls the publishing API. Each agent gets a focused role, goal, and backstory to keep prompts tight.

# agents.py
from crewai import Agent
from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.3,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

strategist = Agent(
    role="Social Media Strategist",
    goal="Select high-engagement topics and angles for the target audience",
    backstory=(
        "You analyze trends, audience personas, and brand voice to pick "
        "topics that drive clicks and conversations. You output a structured "
        "content plan with topic, angle, target platform, and suggested posting time."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

copywriter = Agent(
    role="Social Media Copywriter",
    goal="Write platform-optimized copy that matches brand voice and drives action",
    backstory=(
        "You turn content plans into publish-ready posts. You know character limits, "
        "hashtag conventions, and CTA patterns for X, LinkedIn, and Threads. "
        "You output JSON with platform, copy, hashtags, and media suggestions."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

scheduler = Agent(
    role="Social Media Scheduler",
    goal="Publish approved posts via platform APIs at the scheduled time",
    backstory=(
        "You take finalized copy and media references, call the appropriate "
        "platform API (X, LinkedIn, Threads), and return confirmation with post IDs. "
        "You handle rate limits and retries gracefully."
    ),
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

Step 3: Define tasks with explicit dependencies

Tasks declare context to pass outputs downstream. The strategist produces a ContentPlan; the copywriter consumes it and emits PostDrafts; the scheduler consumes PostDrafts and returns PublishResult. Pydantic models make the contract explicit and validate at runtime.

# tasks.py
from crewai import Task
from pydantic import BaseModel, Field
from typing import List, Literal, Optional
from agents import strategist, copywriter, scheduler

class ContentPlan(BaseModel):
    topic: str
    angle: str
    target_platforms: List[Literal["x", "linkedin", "threads"]]
    suggested_time_utc: str  # ISO 8601
    key_messages: List[str]
    target_audience: str

class PostDraft(BaseModel):
    platform: Literal["x", "linkedin", "threads"]
    copy: str
    hashtags: List[str]
    media_urls: List[str] = Field(default_factory=list)
    scheduled_time_utc: str

class PostDrafts(BaseModel):
    drafts: List[PostDraft]

class PublishResult(BaseModel):
    platform: str
    post_id: str
    published_at_utc: str
    status: Literal["published", "failed"]
    error: Optional[str] = None

plan_task = Task(
    description=(
        "Given the brand context and campaign goal, produce a ContentPlan "
        "with 1-3 topics, each with angle, target platforms, suggested time, "
        "key messages, and target audience. Output must validate against the ContentPlan schema."
    ),
    expected_output="A single ContentPlan JSON object",
    agent=strategist,
    output_json=ContentPlan,
)

draft_task = Task(
    description=(
        "Using the ContentPlan from the strategist, write platform-optimized "
        "copy for each target platform. Respect character limits (X: 280, "
        "LinkedIn: 3000, Threads: 500), include 3-5 relevant hashtags, "
        "and suggest media URLs if applicable. Output must validate against PostDrafts schema."
    ),
    expected_output="A PostDrafts JSON object containing one draft per platform",
    agent=copywriter,
    context=[plan_task],
    output_json=PostDrafts,
)

publish_task = Task(
    description=(
        "For each PostDraft, call the appropriate platform publishing API. "
        "Return a list of PublishResult objects with post_id, published_at_utc, "
        "and status. If a platform API returns rate limit, retry with exponential "
        "backoff up to 3 times before marking failed."
    ),
    expected_output="A list of PublishResult JSON objects",
    agent=scheduler,
    context=[draft_task],
    output_json=List[PublishResult],
)

Step 4: Assemble the crew with memory and tools

Enable memory=True so agents retain context across runs — useful for multi-day campaigns. Attach custom tools for the scheduler to call real APIs. Here we stub the publishing tool; replace with actual HTTP clients for X (v2), LinkedIn (UGS), and Threads (Graph API).

# tools.py
from crewai_tools import BaseTool
from pydantic import BaseModel, Field
from typing import Literal
import httpx
import os
import time

class PublishInput(BaseModel):
    platform: Literal["x", "linkedin", "threads"]
    copy: str
    media_urls: List[str] = Field(default_factory=list)

class PublishTool(BaseTool):
    name: str = "publish_post"
    args_schema: type[BaseModel] = PublishInput

    def _run(self, platform: str, copy: str, media_urls: list = None) -> dict:
        media_urls = media_urls or []
        # Stub — replace with real API calls
        if platform == "x":
            return self._publish_x(copy, media_urls)
        elif platform == "linkedin":
            return self._publish_linkedin(copy, media_urls)
        elif platform == "threads":
            return self._publish_threads(copy, media_urls)
        raise ValueError(f"Unknown platform: {platform}")

    def _publish_x(self, copy: str, media_urls: list) -> dict:
        # POST https://api.twitter.com/2/tweets
        # Requires OAuth 2.0 Bearer token with tweet.write scope
        return {"post_id": "x_12345", "published_at_utc": "2025-01-15T14:30:00Z", "status": "published"}

    def _publish_linkedin(self, copy: str, media_urls: list) -> dict:
        # POST https://api.linkedin.com/rest/posts
        # Requires rw_organization_admin or w_member_social
        return {"post_id": "li_67890", "published_at_utc": "2025-01-15T14:30:00Z", "status": "published"}

    def _publish_threads(self, copy: str, media_urls: list) -> dict:
        # POST https://graph.threads.net/v1.0/me/threads
        # Requires threads_basic, threads_content_publish
        return {"post_id": "th_abcde", "published_at_utc": "2025-01-15T14:30:00Z", "status": "published"}

publish_tool = PublishTool()

Now wire the crew:

# crew.py
from crewai import Crew, Process
from tasks import plan_task, draft_task, publish_task
from tools import publish_tool

# Attach tool to scheduler task
publish_task.tools = [publish_tool]

crew = Crew(
    agents=[plan_task.agent, draft_task.agent, publish_task.agent],
    tasks=[plan_task, draft_task, publish_task],
    process=Process.sequential,
    memory=True,
    verbose=True,
    max_rpm=60,  # Respect provider rate limits
)

Step 5: Run the crew with sample input

Create a driver script that loads brand context, kicks off the crew, and prints structured results. The input can come from a JSON file, a database, or an API — here we use a dict for clarity.

# run_crew.py
import json
from crew import crew

brand_context = {
    "brand_name": "DevTools Inc",
    "voice": "Technical, concise, slightly witty. No fluff.",
    "audience": "Backend engineers and DevOps practitioners",
    "campaign_goal": "Launch new open-source CLI for database migrations",
    "key_links": {
        "github": "https://github.com/devtools/migrate",
        "docs": "https://migrate.devtools.io",
    },
    "avoid": ["Marketing buzzwords", "Exclamation marks", "Vague claims"],
}

inputs = {
    "brand_context": json.dumps(brand_context, indent=2),
    "campaign_goal": brand_context["campaign_goal"],
}

result = crew.kickoff(inputs=inputs)

print("\n=== CREW RESULT ===")
print(result.raw)

Run it:

python run_crew.py

You should see verbose logs from each agent, then a final JSON block containing the PublishResult list. Capture that output — it’s your verification artifact.

Step 6: Verify success at each stage

Don’t trust the final output alone. Add checkpoints that validate intermediate artifacts before the next agent runs. This catches hallucinated hashtags, character-limit violations, and malformed API payloads early.

# verify.py
from pydantic import ValidationError
from tasks import ContentPlan, PostDrafts, PublishResult
from typing import List
import json

def verify_plan(raw: str) -> ContentPlan:
    try:
        plan = ContentPlan.model_validate_json(raw)
    except ValidationError as e:
        raise ValueError(f"ContentPlan validation failed: {e}")
    # Business rules
    assert 1 <= len(plan.target_platforms) <= 3, "Must target 1-3 platforms"
    assert plan.key_messages, "At least one key message required"
    return plan

def verify_drafts(raw: str, plan: ContentPlan) -> PostDrafts:
    try:
        drafts = PostDrafts.model_validate_json(raw)
    except ValidationError as e:
        raise ValueError(f"PostDrafts validation failed: {e}")
    # Ensure each platform in plan has a draft
    drafted_platforms = {d.platform for d in drafts.drafts}
    missing = set(plan.target_platforms) - drafted_platforms
    assert not missing, f"Missing drafts for platforms: {missing}"
    # Character limits
    limits = {"x": 280, "linkedin": 3000, "threads": 500}
    for draft in drafts.drafts:
        limit = limits[draft.platform]
        assert len(draft.copy) <= limit, f"{draft.platform} copy exceeds {limit} chars"
        assert 3 <= len(draft.hashtags) <= 5, f"{draft.platform} needs 3-5 hashtags"
    return drafts

def verify_publish(raw: str) -> List[PublishResult]:
    try:
        results = [PublishResult.model_validate_json(raw)] if raw.strip().startswith("{") \
                  else [PublishResult.model_validate(obj) for obj in json.loads(raw)]
    except (ValidationError, json.JSONDecodeError) as e:
        raise ValueError(f"PublishResult validation failed: {e}")
    failed = [r for r in results if r.status == "failed"]
    if failed:
        print(f"WARNING: {len(failed)} posts failed to publish")
        for f in failed:
            print(f"  {f.platform}: {f.error}")
    return results

Hook these into your driver:

# run_crew.py (updated)
import json
from crew import crew
from verify import verify_plan, verify_drafts, verify_publish

brand_context = { ... }  # same as before
inputs = {"brand_context": json.dumps(brand_context, indent=2), "campaign_goal": brand_context["campaign_goal"]}

# Kick off and capture intermediate outputs via crew's task outputs
result = crew.kickoff(inputs=inputs)

# The crew stores each task's output in task.output.raw
plan_raw = crew.tasks[0].output.raw
drafts_raw = crew.tasks[1].output.raw
publish_raw = crew.tasks[2].output.raw

plan = verify_plan(plan_raw)
drafts = verify_drafts(drafts_raw, plan)
publish_results = verify_publish(publish_raw)

print("\n=== VERIFIED RESULTS ===")
print(f"Plan: {plan.topic} -> {plan.target_platforms}")
print(f"Drafts: {len(drafts.drafts)} platforms")
print(f"Published: {sum(1 for r in publish_results if r.status == 'published')}/{len(publish_results)}")

Run again. You should see clean verification output with zero assertion errors.

Step 7: Production hardening

A crew that works locally needs guardrails before it schedules real posts. Address these before deploying:

Rate limiting and retries — The PublishTool stub returns success instantly. Real APIs return 429. Implement exponential backoff with jitter, and honor Retry-After headers. Set max_rpm on the crew to stay within your provider’s quota.

Idempotency — Generate a deterministic idempotency_key per post (hash of platform + copy + scheduled time) and pass it to the API if supported. This prevents duplicate posts on retry.

Approval gate — Insert a human-in-the-loop task between drafting and publishing. CrewAI supports human_input=True on tasks; use it to pause for review in a Slack message, email, or internal dashboard.

Secrets management — Never hardcode API tokens. Load from a vault (AWS Secrets Manager, HashiCorp Vault, or 1Password CLI) at runtime. Rotate tokens on a schedule.

Observability — Emit structured logs (JSON) with crew_run_id, task_name, agent_role, duration_ms, token_usage, and status. Ship to your logging stack. Track cost per run by summing prompt_tokens + completion_tokens across agents.

Fallback routing — If your primary LLM provider degrades, route to a backup. A gateway that forwards provider cache-control hints and honors client routing directives lets you swap models without code changes — useful when you need lower latency for the scheduler agent versus higher reasoning for the strategist.

Testing — Unit-test each tool in isolation with mocked HTTP responses. Integration-test the full crew against a staging environment with test accounts on each platform. Assert that published posts appear in the platform UI and match the approved copy.

Extending the crew

This three-agent pattern scales. Add a media_generator agent that creates images via DALL·E or Midjourney and uploads to your CDN. Add an analytics_collector that runs 24h post-publish to fetch impressions, engagement, and CTR — then feeds results back to the strategist for the next cycle. The crew becomes a closed loop: plan → create → publish → measure → replan.

The crewai social media scheduling crew example here is deliberately minimal but structurally sound. Every component — agents, tasks, tools, verification — is replaceable without rewriting the orchestration. That’s the point: build the skeleton once, swap organs as requirements change.