How to Build a Multi-Agent SEO Content Pipeline with Claude API

ANTHROPIC_API_KEY=your_api_key_here

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

# Shared model identifier used by all agents in the pipeline
MODEL = "claude-sonnet-4-5"

# Single Anthropic client instance shared across all agents
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

MAX_TOKENS = 4096

import json
from config import client, MODEL, MAX_TOKENS


class ResearcherAgent:
    """
    Agent 1: Generates a structured SEO research brief from a topic string.
    Outputs JSON with keywords, audience profile, pain points, and content angle.
    """

    def __init__(self):
        self.system_prompt = """You are an expert SEO content strategist with 10 years of experience
in keyword research and audience analysis. Your job is to produce structured research briefs
that writers use to create high-ranking blog content.

When given a content topic, you MUST respond with a valid JSON object containing:
- primary_keyword: the single best target keyword phrase (string)
- secondary_keywords: list of 4-6 supporting keyword phrases (list of strings)
- target_audience: description of who will read this content (string)
- pain_points: list of 3-5 specific problems this audience faces (list of strings)
- key_talking_points: list of 5-7 concrete points the article must cover (list of strings)
- recommended_word_count: ideal article length as an integer (integer)
- content_angle: the unique angle or hook that makes this article stand out (string)
- suggested_h2_headings: list of 4-6 suggested section headings (list of strings)

Return ONLY the JSON object. No preamble, no explanation, no markdown fences."""

        # Tool definition for structured research output
        self.tools = [
            {
                "name": "generate_research_brief",
                "description": "Generate a structured SEO research brief for a given content topic",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "primary_keyword": {
                            "type": "string",
                            "description": "The main target keyword phrase for the article"
                        },
                        "secondary_keywords": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Supporting keyword phrases to include naturally"
                        },
                        "target_audience": {
                            "type": "string",
                            "description": "Description of the primary reader persona"
                        },
                        "pain_points": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Specific problems the audience is trying to solve"
                        },
                        "key_talking_points": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Core points the article must cover"
                        },
                        "recommended_word_count": {
                            "type": "integer",
                            "description": "Recommended article length in words"
                        },
                        "content_angle": {
                            "type": "string",
                            "description": "The unique hook or angle that differentiates this article"
                        },
                        "suggested_h2_headings": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "Suggested H2 section headings for the article structure"
                        }
                    },
                    "required": [
                        "primary_keyword",
                        "secondary_keywords",
                        "target_audience",
                        "pain_points",
                        "key_talking_points",
                        "recommended_word_count",
                        "content_angle",
                        "suggested_h2_headings"
                    ]
                }
            }
        ]

    def run(self, topic: str) -> dict:
        """
        Takes a topic string and returns a structured research brief as a dict.
        Uses Claude's tool use feature to enforce structured JSON output.
        """
        print(f"\n[Researcher Agent] Analyzing topic: '{topic}'")

        messages = [
            {
                "role": "user",
                "content": f"Create a comprehensive SEO research brief for this topic: {topic}"
            }
        ]

        response = client.messages.create(
            model=MODEL,
            max_tokens=MAX_TOKENS,
            system=self.system_prompt,
            tools=self.tools,
            # Force Claude to use the tool for guaranteed structured output
            tool_choice={"type": "tool", "name": "generate_research_brief"},
            messages=messages
        )

        # Extract the tool use block from the response
        for block in response.content:
            if block.type == "tool_use" and block.name == "generate_research_brief":
                research_brief = block.input
                print(f"[Researcher Agent] Brief generated. Primary keyword: '{research_brief['primary_keyword']}'")
                return research_brief

        # Fallback: should never hit this if tool_choice is set correctly
        raise ValueError("Researcher Agent failed to return a structured brief.")

from config import client, MODEL, MAX_TOKENS


class SEOOptimizerAgent:
    """
    Agent 2: Writes a full SEO-optimized blog post from a research brief.
    Applies on-page SEO rules including keyword placement, heading structure,
    and internal linking guidance.
    """

    def __init__(self):
        self.system_prompt = """You are a professional SEO content writer who produces
high-ranking, human-sounding blog posts. You never write generic filler content.
Every paragraph adds specific value.

Follow these SEO optimization rules exactly:
1. Include the primary keyword in the first 100 words of the introduction
2. Use the primary keyword 3-5 times total throughout the article (natural density)
3. Include at least 3 secondary keywords, each used 1-2 times
4. Write an H1 title that includes the primary keyword
5. Use H2 headings for major sections (use the suggested headings from the brief)
6. Use H3 subheadings where it improves scannability
7. Write a meta description under 160 characters that includes the primary keyword
8. Include a clear call-to-action in the final section
9. Keep paragraphs to 3 sentences or fewer
10. Write in second person ("you") to speak directly to the reader

Structure your response exactly like this:
META_DESCRIPTION: [your meta description here]
---
[Full article starting with H1]"""

    def run(self, research_brief: dict) -> str:
        """
        Takes a research brief dict and returns a full SEO-optimized blog post as a string.
        """
        print(f"\n[SEO Optimizer Agent] Writing article for keyword: '{research_brief['primary_keyword']}'")

        # Format the research brief into a clear prompt for the writer agent
        brief_formatted = f"""
Use this research brief to write the blog post:

PRIMARY KEYWORD: {research_brief['primary_keyword']}
SECONDARY KEYWORDS: {', '.join(research_brief['secondary_keywords'])}
TARGET AUDIENCE: {research_brief['target_audience']}
PAIN POINTS TO ADDRESS:
{chr(10).join(f'- {point}' for point in research_brief['pain_points'])}
KEY TALKING POINTS:
{chr(10).join(f'- {point}' for point in research_brief['key_talking_points'])}
TARGET WORD COUNT: {research_brief['recommended_word_count']} words
CONTENT ANGLE: {research_brief['content_angle']}
SUGGESTED H2 HEADINGS:
{chr(10).join(f'- {heading}' for heading in research_brief['suggested_h2_headings'])}

Write the complete blog post now. Follow all SEO optimization rules from your instructions.
"""

        messages = [
            {
                "role": "user",
                "content": brief_formatted
            }
        ]

        response = client.messages.create(
            model=MODEL,
            max_tokens=MAX_TOKENS,
            system=self.system_prompt,
            messages=messages
        )

        article_content = response.content[0].text
        word_count = len(article_content.split())
        print(f"[SEO Optimizer Agent] Article written. Word count: ~{word_count}")

        return article_content

from config import client, MODEL, MAX_TOKENS


class QualityReviewerAgent:
    """
    Agent 3: Evaluates a finished article against an SEO and quality rubric.
    Returns a structured review with scores, pass/fail verdict, and revision notes.
    """

    def __init__(self):
        self.system_prompt = """You are a senior content editor and SEO auditor.
Your job is to evaluate blog posts against a strict quality rubric and provide
actionable, specific feedback.

Evaluate each article on these criteria (score each 1-10):
1. keyword_usage: Primary keyword appears naturally 3-5 times, not stuffed
2. heading_structure: Logical H1 > H2 > H3 hierarchy, headings are descriptive
3. readability: Short paragraphs, clear sentences, no jargon overload
4. content_value: Specific, useful information — not generic filler
5. cta_presence: Clear, compelling call-to-action in the final section
6. meta_description: Present, under 160 chars, includes primary keyword

Scoring:
- 8-10: Excellent, no changes needed
- 5-7: Acceptable, minor improvements suggested
- 1-4: Needs significant revision

PASS threshold: All scores must be 6 or higher AND average score must be 7 or higher.

Respond ONLY with this exact JSON structure:
{
  "scores": {
    "keyword_usage": ,
    "heading_structure": ,
    "readability": ,
    "content_value": ,
    "cta_presence": ,
    "meta_description": 
  },
  "average_score": ,
  "verdict": "PASS" or "FAIL",
  "revision_notes": [list of specific, actionable revision suggestions],
  "strongest_section": "",
  "weakest_section": ""
}"""

    def run(self, article_content: str, research_brief: dict) -> dict:
        """
        Evaluates the article against the research brief and returns a review dict.
        """
        print(f"\n[Quality Reviewer Agent] Reviewing article...")

        review_prompt = f"""Review this article against the following research brief context:

PRIMARY KEYWORD: {research_brief['primary_keyword']}
TARGET AUDIENCE: {research_brief['target_audience']}

ARTICLE TO REVIEW:
{article_content}

Apply the evaluation rubric from your instructions and return only the JSON review object."""

        messages = [
            {
                "role": "user",
                "content": review_prompt
            }
        ]

        response = client.messages.create(
            model=MODEL,
            max_tokens=MAX_TOKENS,
            system=self.system_prompt,
            messages=messages
        )

        raw_response = response.content[0].text.strip()

        # Strip markdown code fences if Claude wraps the JSON in them
        if raw_response.startswith("```"):
            lines = raw_response.split("\n")
            # Remove first and last lines (the ``` fences)
            raw_response = "\n".join(lines[1:-1])

        import json
        review = json.loads(raw_response)

        verdict = review.get("verdict", "UNKNOWN")
        avg = review.get("average_score", 0)
        print(f"[Quality Reviewer Agent] Verdict: {verdict} | Average Score: {avg}/10")

        return review

import json
import time
from researcher_agent import ResearcherAgent
from seo_optimizer_agent import SEOOptimizerAgent
from quality_reviewer_agent import QualityReviewerAgent


class ContentPipelineOrchestrator:
    """
    Orchestrates the 3-agent SEO content pipeline.
    Runs: Researcher -> SEO Optimizer -> Quality Reviewer
    Retries the optimizer once if the reviewer returns a FAIL verdict.
    """

    def __init__(self, max_retries: int = 2):
        self.researcher = ResearcherAgent()
        self.optimizer = SEOOptimizerAgent()
        self.reviewer = QualityReviewerAgent()
        self.max_retries = max_retries

    def run(self, topic: str) -> dict:
        """
        Full pipeline run from topic string to reviewed article.
        Returns a results dict with the research brief, article, and review.
        """
        pipeline_start = time.time()
        print(f"\n{'='*60}")
        print(f"PIPELINE START: {topic}")
        print(f"{'='*60}")

        # --- Agent 1: Research ---
        research_brief = self.researcher.run(topic)

        # --- Agent 2: Write (with retry loop) ---
        article_content = None
        review = None
        attempt = 0

        while attempt < self.max_retries:
            attempt += 1
            print(f"\n[Orchestrator] Writing attempt {attempt} of {self.max_retries}")

            # Pass the full research brief to the optimizer
            article_content = self.optimizer.run(research_brief)

            # --- Agent 3: Review ---
            review = self.reviewer.run(article_content, research_brief)

            if review.get("verdict") == "PASS":
                print(f"\n[Orchestrator] Article PASSED review on attempt {attempt}.")
                break
            else:
                print(f"\n[Orchestrator] Article FAILED review. Revision notes:")
                for note in review.get("revision_notes", []):
                    print(f"  - {note}")

                if attempt < self.max_retries:
                    # Inject reviewer feedback into the research brief for the next attempt
                    revision_context = "\n".join(review.get("revision_notes", []))
                    research_brief["revision_notes"] = revision_context
                    print(f"[Orchestrator] Retrying with revision feedback injected...")
                else:
                    print(f"[Orchestrator] Max retries reached. Returning best attempt.")

        elapsed = round(time.time() - pipeline_start, 2)
        print(f"\n{'='*60}")
        print(f"PIPELINE COMPLETE in {elapsed}s")
        print(f"{'='*60}\n")

        return {
            "topic": topic,
            "research_brief": research_brief,
            "article": article_content,
            "review": review,
            "elapsed_seconds": elapsed
        }

    def save_output(self, results: dict, filename: str = "output.json") -> None:
        """Saves the full pipeline results to a JSON file for inspection."""
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(results, f, indent=2, ensure_ascii=False)
        print(f"[Orchestrator