Designing Agentic Marketing on AWS: A Bedrock AgentCore Blueprint for GEO and AEO

"""
Agentic Marketing Supervisor on Amazon Bedrock AgentCore
========================================================
pip install bedrock-agentcore strands-agents boto3
"""
import boto3
from pathlib import Path
from bedrock_agentcore import AgentCoreApp, MemoryClient
from strands import Agent, tool
from strands.models import BedrockModel

# Load SEO/GEO skill files from https://github.com/aaron-he-zhu/seo-geo-claude-skills
SKILLS_ROOT = Path("./skills")

def load_skills(*names: str) -> str:
    return "".join(
        f"\n<skill name='{n}'>\n{(SKILLS_ROOT / n / 'SKILL.md').read_text()}\n</skill>\n"
        for n in names
    )

# MCP tools exposed through AgentCore Gateway
gateway = boto3.client("bedrock-agentcore-gateway")

@tool
def keyword_research(seed_topic: str) -> dict:
    """Pull keyword + AI-prompt data and which LLMs cite competitors."""
    return gateway.invoke_tool(
        gateway_name="seo-data-gateway",
        tool_name="keyword_research",
        payload={"seed": seed_topic},
    )

@tool
def check_ai_citations(url: str, queries: list[str]) -> dict:
    """Query ChatGPT, Perplexity, Claude, Gemini, Google AI Overviews."""
    return gateway.invoke_tool(
        gateway_name="seo-data-gateway",
        tool_name="ai_citation_check",
        payload={"url": url, "queries": queries},
    )

@tool
def run_quality_audit(draft_s3_uri: str) -> dict:
    """Run CORE-EEAT (80) + CITE (40). Returns score 0-120 + deltas."""
    return gateway.invoke_tool(
        gateway_name="audit-gateway",
        tool_name="full_audit",
        payload={"draft": draft_s3_uri},
    )

@tool
def publish_to_cms(draft_s3_uri: str, schema_jsonld: dict, slug: str) -> dict:
    return gateway.invoke_tool(
        gateway_name="cms-gateway",
        tool_name="publish",
        payload={"draft": draft_s3_uri, "schema": schema_jsonld, "slug": slug},
    )

# Specialist sub-agents — each loads only the skills it needs
model = BedrockModel(
    model_id="anthropic.claude-sonnet-4-5-20251022-v1:0",
    region_name="us-east-1",
    temperature=0.3,
)

researcher = Agent(
    model=model,
    system_prompt="Find GEO citation gaps where LLMs cite competitors but not us."
        + load_skills("research/keyword-research",
                      "research/serp-analysis",
                      "research/content-gap-analysis"),
    tools=[keyword_research, check_ai_citations],
)

writer = Agent(
    model=model,
    system_prompt="Write content LLMs cite. Extractable facts, expert quotes, "
                  "statistics, JSON-LD schema alongside every draft."
        + load_skills("build/seo-content-writer",
                      "build/geo-content-optimizer",
                      "build/schema-markup-generator"),
)

auditor = Agent(
    model=model,
    system_prompt="Strict E-E-A-T auditor. Block publish if score < 95."
        + load_skills("cross-cutting/content-quality-auditor",
                      "cross-cutting/domain-authority-auditor"),
    tools=[run_quality_audit],
)

# Supervisor delegates and persists handoffs to long-term memory
memory = MemoryClient(memory_id="marketing-agent-memory")

@tool
def delegate_research(brief: str) -> str: return researcher(brief).message
@tool
def delegate_writing(brief: str)  -> str: return writer(brief).message
@tool
def delegate_audit(uri: str)      -> str: return auditor(f"Audit: {uri}").message

supervisor = Agent(
    model=model,
    system_prompt=(
        "You orchestrate a GEO/AEO workflow. Each run:\n"
        "1. Read prior handoff from memory.\n"
        "2. Identify highest-ROI citation gap via researcher.\n"
        "3. Commission draft via writer.\n"
        "4. Gate publish on auditor (score >= 95).\n"
        "5. Write handoff summary to memory before exiting.\n"
        + load_skills("cross-cutting/memory-management",
                      "cross-cutting/entity-optimizer")
    ),
    tools=[delegate_research, delegate_writing, delegate_audit, publish_to_cms],
)

# AgentCore Runtime entrypoint — EventBridge invokes this daily
app = AgentCoreApp()

@app.entrypoint
def handler(payload: dict) -> dict:
    prior = memory.retrieve(namespace="marketing/handoffs",
                            query="latest weekly handoff", top_k=1)
    context = prior[0]["text"] if prior else "First run."

    goal = payload.get("goal",
        "Find and fill the highest-ROI GEO citation gap this week.")

    result = supervisor(f"Prior handoff:\n{context}\n\nGoal:\n{goal}")

    memory.store(namespace="marketing/handoffs", content=result.message)
    return {"status": "ok", "summary": result.message}

if __name__ == "__main__":
    app.run()

agentcore deploy --name marketing-supervisor --entrypoint app.handler

aws events put-rule --name daily-marketing-agent \
  --schedule-expression "cron(0 11 * * ? *)"