How to Build an AI Agent for Social Media

from dataclasses import dataclass, field

@dataclass
class Campaign:
    goal: str
    audience: str
    start: str                       # ISO date, e.g. "2026-07-01"
    days: int
    platforms: list[str]             # ["linkedin", "instagram", "x"]
    voice: str                       # tone + do/do-not words + examples
    campaign_id: str = ""

def load_campaign(brief: dict, brand_voice: str) -> Campaign:
    c = Campaign(goal=brief["goal"], audience=brief["audience"],
                 start=brief["start"], days=brief["days"],
                 platforms=brief["platforms"], voice=brand_voice,
                 campaign_id=brief["id"])     # id makes the run idempotent
    return c
# One structured object feeds every later step; no model call needed here.

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class Post(BaseModel):
    topic: str; date: str; platform: str

class Calendar(BaseModel):
    posts: list[Post]

def plan(c) -> Calendar:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Plan {c.days} days of posts for {c.platforms} on goal "
                  f"'{c.goal}' for {c.audience}, starting {c.start}."],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Calendar))
    return res.parsed                # a typed Calendar object

from google import genai
client = genai.Client()
RULES = {"linkedin": "professional, 1 to 2 short paragraphs, 0 to 2 hashtags",
         "instagram": "warm and visual, 3 to 8 relevant hashtags",
         "x": "punchy, under 280 characters, 1 to 2 hashtags"}

def draft_post(topic: str, platform: str, voice: str) -> str:
    return client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Brand voice: {voice}\nWrite a {platform} post about "
                  f"'{topic}'. Style: {RULES[platform]}. It is for review, "
                  f"not to be published automatically."]).text
# The return value is a draft shown to a person, never published by the agent.

def attach_visual(post: dict, asset_library, generate: bool = False) -> dict:
    # Default: write an image brief and try the approved asset library by tag.
    post["image_brief"] = f"Visual for: {post['topic']}, on-brand, {post['platform']}"
    match = asset_library.find(tags=[post["topic"], post["platform"]])
    if match:
        post["image"] = match.url            # reuse an approved asset
        post["image_source"] = "library"
    elif generate:                           # only if you opted in
        post["image"] = image_model.create(post["image_brief"])
        post["image_source"] = "generated"   # labelled clearly for review
    else:
        post["image_source"] = "brief_only"  # a person sources the image
    return post
# Captions plus an image brief by default; generation only when you turn it on.

def schedule(post: dict, account) -> dict:
    # post is already approved; this only places it on the calendar.
    ref = scheduler.create(
        account_id=account.id,
        platform=post["platform"],
        text=post["text"],
        media=post.get("image"),
        publish_at=post["date"])             # the planned time
    post["scheduled_ref"] = ref
    ledger.record(post, status="scheduled")
    return post
# A queued, approved post; nothing is published until a person approves below.

def queue_for_approval(post: dict) -> None:
    approval_queue.add(post)                 # awaits a human; never auto-posts

def on_approval(post: dict, final_text: str) -> None:
    publisher.publish(platform=post["platform"], text=final_text,
                      media=post.get("image"))       # only after approval
    ledger.record(post, status="published")
    audit.log("published", post["id"], final_text)
# The agent drafts, schedules, and queues; publishing happens only on approval.

def report(campaign_id: str) -> dict:
    rows = []
    for post in ledger.published(campaign_id):
        m = insights.fetch(post["platform"], post["scheduled_ref"])
        rows.append({"topic": post["topic"], "platform": post["platform"],
                     "reach": m["reach"], "likes": m["likes"],
                     "clicks": m.get("clicks", 0)})
    summary = aggregate_by_topic(rows)       # which topics and networks won
    ledger.save_report(campaign_id, summary) # context for the next plan
    return summary
# Metrics pulled per post, aggregated by topic, and fed into the next calendar.

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

social_pipeline = SequentialAgent(name="social_pipeline",
    sub_agents=[planner, drafter, reporter])   # each an LlmAgent with the tools

runner = Runner(agent=social_pipeline, app_name="social",
                session_service=InMemorySessionService())

def guard(post: dict) -> str | None:
    if store.seen(post["campaign_id"]):   return "already_planned"  # idempotent
    if not on_brand(post["text"]):        return "needs_review"     # brand check
    if rate_limited(post["platform"]):    return "wait"             # rate limit
    return None
# Run per campaign via cron / Cloud Scheduler. Publishing always needs approval.