How to Build a Multi-Agent SEO Content Pipeline with Claude API (Step-by-Step)

mkdir seo-agent-pipeline
cd seo-agent-pipeline
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-your-key-here

import os
import anthropic
from dotenv import load_dotenv

load_dotenv()

class BaseAgent:
    """
    Shared foundation for all agents in the pipeline.
    Every agent gets its own Anthropic client and model config.
    """

    MODEL = "claude-sonnet-4-6"
    MAX_TOKENS = 4096

    def __init__(self, name: str, system_prompt: str):
        self.name = name
        self.system_prompt = system_prompt
        self.client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    def run(self, user_message: str) -> str:
        """Send a message and return the text response."""
        response = self.client.messages.create(
            model=self.MODEL,
            max_tokens=self.MAX_TOKENS,
            system=self.system_prompt,
            messages=[{"role": "user", "content": user_message}]
        )
        return response.content[0].text

    def run_with_tools(self, user_message: str, tools: list, messages: list = None) -> anthropic.types.Message:
        """Send a message with tool definitions and return the full response object."""
        if messages is None:
            messages = [{"role": "user", "content": user_message}]

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

import json
from base_agent import BaseAgent

TOPIC_ANALYZER_SYSTEM_PROMPT = """
You are an expert SEO strategist and keyword researcher.
When given a topic, you analyze it for:
- Primary keyword (high volume, low competition)
- 3-5 secondary/LSI keywords
- Search intent (informational, commercial, transactional)
- Recommended content angle and headline
- Target audience persona

Always respond in valid JSON format. Be specific and data-driven.
"""

# Tool definition for the orchestrator to invoke this agent
TOPIC_ANALYSIS_TOOL = {
    "name": "analyze_topic",
    "description": (
        "Analyzes a content topic for SEO potential. Returns keyword strategy, "
        "search intent classification, and a recommended content angle."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "topic": {
                "type": "string",
                "description": "The subject or niche to analyze (e.g., 'Naples FL waterfront homes')"
            },
            "industry": {
                "type": "string",
                "description": "The industry context (e.g., 'real estate', 'healthcare', 'restaurant')"
            }
        },
        "required": ["topic", "industry"]
    }
}


class TopicAnalyzerAgent(BaseAgent):

    def __init__(self):
        super().__init__(
            name="TopicAnalyzer",
            system_prompt=TOPIC_ANALYZER_SYSTEM_PROMPT
        )

    def analyze(self, topic: str, industry: str) -> dict:
        """Run topic analysis and return structured JSON data."""
        prompt = f"""
Analyze this topic for SEO content creation:

Topic: {topic}
Industry: {industry}

Return a JSON object with these exact keys:
- primary_keyword (string)
- secondary_keywords (array of strings)
- search_intent (string: informational | commercial | transactional)
- content_angle (string: the recommended headline or angle)
- target_audience (string: describe who is searching for this)
- estimated_word_count (integer: recommended post length)
"""
        raw = self.run(prompt)

        # Strip markdown code fences if Claude wraps the JSON
        cleaned = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
        return json.loads(cleaned)

from base_agent import BaseAgent

CONTENT_WRITER_SYSTEM_PROMPT = """
You are a senior SEO content writer with deep expertise in local search and
long-form blog content. You write in a clear, conversational tone — like a
knowledgeable friend giving real advice.

Your posts always:
- Open with the reader's problem, not a definition
- Use the primary keyword in the first 100 words
- Include H2 and H3 headings that match how people actually search
- Have paragraphs of 2-3 sentences max
- End with a specific call to action
- Avoid corporate jargon and buzzword stacking

You write complete, publish-ready content — not outlines.
"""

CONTENT_WRITING_TOOL = {
    "name": "write_content",
    "description": (
        "Writes a full SEO-optimized blog post based on a keyword strategy. "
        "Returns complete HTML-structured article content."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "primary_keyword": {
                "type": "string",
                "description": "The main keyword to optimize for"
            },
            "secondary_keywords": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Supporting keywords to weave in naturally"
            },
            "content_angle": {
                "type": "string",
                "description": "The headline or angle to write toward"
            },
            "target_audience": {
                "type": "string",
                "description": "Who is reading this post"
            },
            "word_count": {
                "type": "integer",
                "description": "Target word count for the post"
            },
            "industry": {
                "type": "string",
                "description": "Industry context for tone and examples"
            }
        },
        "required": [
            "primary_keyword", "secondary_keywords",
            "content_angle", "target_audience", "word_count", "industry"
        ]
    }
}


class ContentWriterAgent(BaseAgent):

    def __init__(self):
        super().__init__(
            name="ContentWriter",
            system_prompt=CONTENT_WRITER_SYSTEM_PROMPT
        )

    def write(
        self,
        primary_keyword: str,
        secondary_keywords: list,
        content_angle: str,
        target_audience: str,
        word_count: int,
        industry: str
    ) -> str:
        """Generate the full blog post and return it as a string."""
        keywords_list = ", ".join(secondary_keywords)

        prompt = f"""
Write a complete blog post with these specs:

Primary Keyword: {primary_keyword}
Secondary Keywords: {keywords_list}
Headline / Angle: {content_angle}
Target Audience: {target_audience}
Target Word Count: {word_count} words
Industry: {industry}

Format the post with clear H2 and H3 headings.
Include the primary keyword in the title, first paragraph, one H2, and the conclusion.
End with a CTA encouraging readers to take the next step.
Write the full post now — not an outline.
"""
        return self.run(prompt)

import json
from base_agent import BaseAgent

METADATA_GENERATOR_SYSTEM_PROMPT = """
You are an SEO technical specialist. Given a blog post and its primary keyword,
you generate all necessary on-page metadata and structured data.

You always return valid JSON. Title tags stay under 60 characters.
Meta descriptions stay under 160 characters and include a benefit or hook.
Slugs are lowercase, hyphenated, and keyword-rich.
"""

METADATA_GENERATION_TOOL = {
    "name": "generate_metadata",
    "description": (
        "Generates complete SEO metadata for a finished blog post including "
        "title tag, meta description, URL slug, Open Graph fields, and FAQ schema."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "post_content": {
                "type": "string",
                "description": "The full blog post text"
            },
            "primary_keyword": {
                "type": "string",
                "description": "The main keyword the post targets"
            },
            "industry": {
                "type": "string",
                "description": "Industry for context"
            }
        },
        "required": ["post_content", "primary_keyword", "industry"]
    }
}


class MetadataGeneratorAgent(BaseAgent):

    def __init__(self):
        super().__init__(
            name="MetadataGenerator",
            system_prompt=METADATA_GENERATOR_SYSTEM_PROMPT
        )

    def generate(self, post_content: str, primary_keyword: str, industry: str) -> dict:
        """Generate SEO metadata and return as a structured dict."""
        prompt = f"""
Generate complete SEO metadata for this blog post.

Primary Keyword: {primary_keyword}
Industry: {industry}

Post Content (first 800 chars for context):
{post_content[:800]}

Return a JSON object with these exact keys:
- title_tag (string, max 60 chars, includes primary keyword)
- meta_description (string, max 160 chars, includes benefit)
- url_slug (string, lowercase hyphenated)
- og_title (string, can be slightly catchier than title_tag)
- og_description (string, 1-2 sentences)
- focus_keyword (string)
- secondary_keywords (array of strings)
- reading_time_minutes (integer)
- word_count (integer, estimate from content)
- schema_faq (array of objects with 'question' and 'answer' keys, 3 items)
"""
        raw = self.run(prompt)
        cleaned = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
        return json.loads(cleaned)

import json
import anthropic
from base_agent import BaseAgent
from topic_analyzer import TopicAnalyzerAgent, TOPIC_ANALYSIS_TOOL
from content_writer import ContentWriterAgent, CONTENT_WRITING_TOOL
from metadata_generator import MetadataGeneratorAgent, METADATA_GENERATION_TOOL

ORCHESTRATOR_SYSTEM_PROMPT = """
You are a content pipeline orchestrator. Your job is to coordinate three specialized agents
to produce a complete, SEO-optimized blog post from a single topic input.

Always follow this sequence:
1. Call analyze_topic first to get keyword strategy
2. Use those results to call write_content
3. Pass the written content to generate_metadata
4. Return a final summary with all three outputs combined

Never skip a step. Never fabricate tool results. Always use the actual tool outputs.
"""

# Tool registry — all three agents' tools in one list
ALL_TOOLS = [
    TOPIC_ANALYSIS_TOOL,
    CONTENT_WRITING_TOOL,
    METADATA_GENERATION_TOOL
]


class ContentPipelineOrchestrator(BaseAgent):

    def __init__(self):
        super().__init__(
            name="Orchestrator",
            system_prompt=ORCHESTRATOR_SYSTEM_PROMPT
        )
        # Instantiate each specialized agent
        self.topic_analyzer = TopicAnalyzerAgent()
        self.content_writer = ContentWriterAgent()
        self.metadata_generator = MetadataGeneratorAgent()

        # Map tool names to handler methods
        self.tool_handlers = {
            "analyze_topic": self._handle_analyze_topic,
            "write_content": self._handle_write_content,
            "generate_metadata": self._handle_generate_metadata,
        }

    def _handle_analyze_topic(self, inputs: dict) -> str:
        """Route analyze_topic tool call to the TopicAnalyzerAgent."""
        result = self.topic_analyzer.analyze(
            topic=inputs["topic"],
            industry=inputs["industry"]
        )
        return json.dumps(result)

    def _handle_write_content(self, inputs: dict) -> str:
        """Route write_content tool call to the ContentWriterAgent."""
        result = self.content_writer.write(
            primary_keyword=inputs["primary_keyword"],
            secondary_keywords=inputs["secondary_keywords"],
            content_angle=inputs["content_angle"],
            target_audience=inputs["target_audience"],
            word_count=inputs["word_count"],
            industry=inputs["industry"]
        )
        # Return as JSON string so the orchestrator can process it
        return json.dumps({"post_content": result})

    def _handle_generate_metadata(self, inputs: dict) -> str:
        """Route generate_metadata tool call to the MetadataGeneratorAgent."""
        result = self.metadata_generator.generate(
            post_content=inputs["post_content"],
            primary_keyword=inputs["primary_keyword"],
            industry=inputs["industry"]
        )
        return json.dumps(result)

    def run_pipeline(self, topic: str, industry: str) -> dict:
        """
        Main entry point. Runs the full agentic loop until Claude
        stops requesting tools and returns the final result.
        """
        print(f"\n🚀 Starting pipeline: '{topic}' | Industry: {industry}\n")

        # Seed the conversation with the initial task
        messages = [
            {
                "role": "user",
                "content": (
                    f"Create a complete SEO blog post for the following:\n\n"
                    f"Topic: {topic}\n"
                    f"Industry: {industry}\n\n"
                    f"Use the available tools in sequence to analyze, write, and generate metadata."
                )
            }
        ]

        final_output = {
            "topic": topic,
            "industry": industry,
            "seo_analysis": None,
            "post_content": None,
            "metadata": None,
        }

        # Agentic loop — keep going until stop_reason is not "tool_use"
        while True:
            response = self.client.messages.create(
                model=self.MODEL,
                max_tokens=self.MAX_TOKENS,
                system=self.system_prompt,
                tools=ALL_TOOLS,
                messages=messages
            )

            print(f"  → Stop reason: {response.stop_reason}")

            # Collect all tool results from this response turn
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    tool_name = block.name
                    tool_input = block.input
                    tool_use_id = block.id

                    print(f"  🔧 Tool called: {tool_name}")

                    # Dispatch to the right handler
                    handler = self.tool_handlers.get(tool_name)
                    if handler:
                        result_str = handler(tool_input)
                    else:
                        result_str = json.dumps({"error": f"Unknown tool: {tool_name}"})

                    # Store result for structured output
                    result_data = json.loads(result_str)
                    if tool_name == "analyze_topic":
                        final_output["seo_analysis"] = result_data
                    elif tool_name == "write_content":
                        final_output["post_content"] = result_data.get("post_content")
                    elif tool_name == "generate_metadata":
                        final_output["metadata"] = result_data

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": tool_use_id,
                        "content": result_str
                    })

            # If no tools were called, Claude is done — exit the loop
            if response.stop_reason != "tool_use":
                print("\n✅ Pipeline complete.\n")
                break

            # Append assistant response and all tool results before next turn
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

        return final_output

import json
from orchestrator import ContentPipelineOrchestrator


def print_pipeline_results(result: dict) -> None:
    """Pretty-print the full pipeline output to terminal."""

    print("=" * 60)
    print("📊 SEO ANALYSIS")
    print("=" * 60)
    if result["seo_analysis"]:
        analysis = result["seo_analysis"]
        print(f"  Primary Keyword:    {analysis.get('primary_keyword')}")
        print(f"  Secondary Keywords: {', '.join(analysis.get('secondary_keywords', []))}")
        print(f"  Search Intent:      {analysis.get('search_intent')}")
        print(f"  Content Angle:      {analysis.get('content_angle')}")
        print(f"  Target Audience:    {analysis.get('target_audience')}")
        print(f"