1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# Initialize the search tool
search_tool = SerperDevTool()

# Define the SEO Research Agent
seo_agent = Agent(
    role="SEO Research Specialist",
    goal="Find high-opportunity keywords and analyze competitor strategies",
    backstory="""You are an experienced SEO analyst who specializes in
    keyword research and competitor gap analysis. You focus on finding
    keywords with decent search volume but low competition — the sweet
    spot for small businesses trying to rank organically.""",
    tools=[search_tool],
    verbose=True,
    memory=True,
    max_iter=15
)

# Define a task for the agent
keyword_research_task = Task(
    description="""Research the keyword landscape for a small SaaS business
    that sells project management tools. Find 10 high-opportunity keywords
    that have moderate search volume but low competition. For each keyword,
    provide the estimated monthly search volume, competition level, and a
    suggested content angle.""",
    expected_output="""A structured list of 10 keywords with:
    - Keyword phrase
    - Estimated monthly search volume
    - Competition level (low/medium/high)
    - Suggested content angle for a blog post""",
    agent=seo_agent
)

# Build and run the crew
crew = Crew(
    agents=[seo_agent],
    tasks=[keyword_research_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
import os
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileWriterTool

search_tool = SerperDevTool()
file_writer = FileWriterTool()

# --- Agent Definitions ---

seo_agent = Agent(
    role="SEO Research Specialist",
    goal="Find high-opportunity keywords and create content briefs",
    backstory="""You are an SEO expert who identifies keyword gaps and
    creates actionable content briefs. You focus on long-tail keywords
    that small businesses can realistically rank for.""",
    tools=[search_tool],
    verbose=True,
    memory=True
)

content_agent = Agent(
    role="Content Writer",
    goal="Write engaging, SEO-optimized blog posts based on research briefs",
    backstory="""You are a technical content writer who turns research
    briefs into clear, actionable blog posts. You write for developers
    and technical founders. Your style is direct, no fluff, and packed
    with practical advice. You naturally incorporate target keywords
    without making the content feel forced.""",
    tools=[file_writer],
    verbose=True,
    memory=True
)

social_agent = Agent(
    role="Social Media Manager",
    goal="Create platform-specific social media content from blog posts",
    backstory="""You are a social media strategist who repurposes blog
    content into engaging posts for Twitter/X, LinkedIn, and Instagram.
    You understand the unique format and audience of each platform.
    You write hooks that stop the scroll.""",
    tools=[file_writer],
    verbose=True,
    memory=True
)

# --- Task Definitions ---

seo_task = Task(
    description="""Research keywords for a small business that sells
    invoicing software for freelancers. Find the top 3 keyword
    opportunities and create a content brief for the best one.
    Include: target keyword, secondary keywords, suggested title,
    outline with H2 headings, and target word count.""",
    expected_output="""A detailed content brief with target keyword,
    secondary keywords, title, H2 outline, and word count target.""",
    agent=seo_agent
)

content_task = Task(
    description="""Using the content brief provided, write a complete
    blog post. Follow the outline structure, naturally incorporate
    the target and secondary keywords, and write in a direct,
    practical tone. Include a compelling introduction and a clear
    call-to-action at the end.""",
    expected_output="""A complete blog post in markdown format,
    ready to publish. Save it to output/blog_post.md""",
    agent=content_agent,
    context=[seo_task]  # Gets output from SEO task
)

social_task = Task(
    description="""Based on the blog post, create social media content:
    1. Three tweet/X posts with different hooks (under 280 chars each)
    2. One LinkedIn post (conversational, 150-200 words)
    3. Two Instagram caption options (with relevant hashtags)
    Save all content to output/social_posts.md""",
    expected_output="""Social media content for Twitter/X, LinkedIn,
    and Instagram, saved to output/social_posts.md""",
    agent=social_agent,
    context=[content_task]  # Gets output from content task
)

# --- Build and Run the Crew ---

crew = Crew(
    agents=[seo_agent, content_agent, social_agent],
    tasks=[seo_task, content_task, social_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import requests

class KeywordInput(BaseModel):
    """Input for keyword difficulty tool."""
    keyword: str = Field(..., description="The keyword to check difficulty for")

class KeywordDifficultyTool(BaseTool):
    name: str = "keyword_difficulty_checker"
    description: str = "Checks the ranking difficulty of a keyword and returns a score from 0-100"
    args_schema: Type[BaseModel] = KeywordInput

    def _run(self, keyword: str) -> str:
        # Replace this with your actual SEO API call
        # Example using a hypothetical API
        response = requests.get(
            "https://api.your-seo-tool.com/difficulty",
            params={"keyword": keyword},
            headers={"Authorization": f"Bearer {os.environ['SEO_API_KEY']}"}
        )
        data = response.json()
        return f"Keyword: {keyword}, Difficulty: {data['score']}/100, Volume: {data['volume']}/mo"

1
2
3
4
5
6
7
8
9
10
11
12
13
from crewai import tool

@tool("WordPress Publisher")
def publish_to_wordpress(title: str, content: str, status: str = "draft") -> str:
    """Publishes a blog post to WordPress via REST API."""
    response = requests.post(
        "https://your-site.com/wp-json/wp/v2/posts",
        json={"title": title, "content": content, "status": status},
        headers={"Authorization": f"Bearer {os.environ['WP_API_TOKEN']}"}
    )
    if response.status_code == 201:
        return f"Post published: {response.json()['link']}"
    return f"Error: {response.status_code}"

1
2
3
4
5
6
content_agent = Agent(
    role="Content Writer",
    goal="Write and publish SEO-optimized blog posts",
    tools=[file_writer, publish_to_wordpress],
    # ...
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import schedule
import time
from daily_content_crew import run_crew

def job():
    print("Running daily content crew...")
    result = run_crew()
    # Send yourself a summary via email or Slack
    notify_slack(f"Daily crew completed: {result.raw[:500]}")

# Run every day at 7 AM
schedule.every().day.at("07:00").do(job)

while True:
    schedule.run_pending()
    time.sleep(60)