Building a Social Media Management Agent: Scheduling, Posting, and Engagement Tracking

Managing Social Media at Scale

Maintaining an active social media presence across multiple platforms is a full-time job. You need to create content, schedule posts at optimal times, respond to comments, track which content performs well, and adjust strategy based on analytics. A social media management agent handles the operational load so humans can focus on creative strategy.

This guide builds an agent that connects to platform APIs, schedules content with intelligent timing, tracks engagement metrics, and generates contextual responses to audience interactions.

Defining the Platform Abstraction

Different platforms have different APIs, but the core operations are the same. We define an abstract interface that each platform implements:

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class SocialPost:
    content: str
    platform: str
    media_urls: list[str] = field(default_factory=list)
    scheduled_time: datetime | None = None
    post_id: str = ""
    status: str = "draft"  # draft, scheduled, published, failed

@dataclass
class EngagementMetrics:
    post_id: str
    likes: int = 0
    comments: int = 0
    shares: int = 0
    impressions: int = 0
    click_through_rate: float = 0.0
    engagement_rate: float = 0.0

class PlatformClient(ABC):
    @abstractmethod
    def publish(self, post: SocialPost) -> str:
        """Publish a post and return the post ID."""
        ...

    @abstractmethod
    def get_metrics(self, post_id: str) -> EngagementMetrics:
        """Fetch engagement metrics for a post."""
        ...

    @abstractmethod
    def get_comments(self, post_id: str) -> list[dict]:
        """Fetch comments on a post."""
        ...

    @abstractmethod
    def reply_to_comment(self, post_id: str, comment_id: str, text: str):
        """Reply to a comment."""
        ...

Implementing a LinkedIn Client

Here is a concrete implementation for LinkedIn using their Marketing API:

import httpx

class LinkedInClient(PlatformClient):
    BASE_URL = "https://api.linkedin.com/v2"

    def __init__(self, access_token: str, author_urn: str):
        self.client = httpx.Client(
            headers={"Authorization": f"Bearer {access_token}"},
            timeout=30,
        )
        self.author_urn = author_urn

    def publish(self, post: SocialPost) -> str:
        payload = {
            "author": self.author_urn,
            "lifecycleState": "PUBLISHED",
            "specificContent": {
                "com.linkedin.ugc.ShareContent": {
                    "shareCommentary": {"text": post.content},
                    "shareMediaCategory": "NONE",
                }
            },
            "visibility": {
                "com.linkedin.ugc.MemberNetworkVisibility": "PUBLIC"
            },
        }
        response = self.client.post(f"{self.BASE_URL}/ugcPosts", json=payload)
        response.raise_for_status()
        return response.json()["id"]

    def get_metrics(self, post_id: str) -> EngagementMetrics:
        response = self.client.get(
            f"{self.BASE_URL}/socialActions/{post_id}",
        )
        data = response.json()
        return EngagementMetrics(
            post_id=post_id,
            likes=data.get("likesSummary", {}).get("totalLikes", 0),
            comments=data.get("commentsSummary", {}).get("totalFirstLevelComments", 0),
        )

    def get_comments(self, post_id: str) -> list[dict]:
        response = self.client.get(
            f"{self.BASE_URL}/socialActions/{post_id}/comments",
        )
        return response.json().get("elements", [])

    def reply_to_comment(self, post_id: str, comment_id: str, text: str):
        self.client.post(
            f"{self.BASE_URL}/socialActions/{post_id}/comments",
            json={
                "parentComment": comment_id,
                "actor": self.author_urn,
                "message": {"text": text},
            },
        )

Intelligent Content Scheduling

The agent determines the best posting times based on historical engagement data rather than generic best-practice charts:

from collections import defaultdict

def analyze_optimal_times(
    metrics_history: list[dict],
) -> dict[str, list[int]]:
    """Analyze historical engagement to find optimal posting hours by day."""
    day_hour_engagement = defaultdict(lambda: defaultdict(list))

    for entry in metrics_history:
        posted_at = datetime.fromisoformat(entry["posted_at"])
        day_name = posted_at.strftime("%A")
        hour = posted_at.hour
        rate = entry.get("engagement_rate", 0)
        day_hour_engagement[day_name][hour].append(rate)

    optimal = {}
    for day, hours in day_hour_engagement.items():
        avg_by_hour = {h: sum(rates) / len(rates) for h, rates in hours.items()}
        sorted_hours = sorted(avg_by_hour, key=avg_by_hour.get, reverse=True)
        optimal[day] = sorted_hours[:3]  # Top 3 hours per day

    return optimal

def schedule_post(
    post: SocialPost,
    optimal_times: dict[str, list[int]],
    target_day: str | None = None,
) -> SocialPost:
    """Schedule a post at the next optimal time."""
    from datetime import timedelta
    import pytz

    now = datetime.now(pytz.utc)

    if target_day and target_day in optimal_times:
        best_hour = optimal_times[target_day][0]
    else:
        today = now.strftime("%A")
        best_hour = optimal_times.get(today, [10])[0]

    scheduled = now.replace(hour=best_hour, minute=0, second=0)
    if scheduled <= now:
        scheduled += timedelta(days=1)

    post.scheduled_time = scheduled
    post.status = "scheduled"
    return post

AI-Powered Engagement Responses

The agent generates contextual replies to comments that match the brand voice:

from openai import OpenAI

llm = OpenAI()

def generate_comment_reply(
    post_content: str,
    comment_text: str,
    brand_voice: str = "professional and friendly",
) -> str:
    """Generate a contextual reply to a social media comment."""
    response = llm.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.5,
        messages=[
            {
                "role": "system",
                "content": (
                    f"You manage social media replies. Voice: {brand_voice}. "
                    "Keep replies concise (1-2 sentences). Be genuine and helpful. "
                    "Never be defensive. If the comment is negative, acknowledge "
                    "the concern and offer to help via DM."
                ),
            },
            {
                "role": "user",
                "content": (
                    f"Original post: {post_content}\n\n"
                    f"Comment: {comment_text}\n\n"
                    "Draft a reply."
                ),
            },
        ],
    )
    return response.choices[0].message.content

def auto_respond_to_comments(
    platform: PlatformClient,
    post: SocialPost,
    brand_voice: str = "professional and friendly",
):
    """Fetch new comments and auto-generate replies for review."""
    comments = platform.get_comments(post.post_id)
    replies = []
    for comment in comments:
        reply = generate_comment_reply(
            post.content, comment.get("text", ""), brand_voice
        )
        replies.append({
            "comment_id": comment.get("id"),
            "comment_text": comment.get("text"),
            "suggested_reply": reply,
            "auto_approved": False,  # Require human approval
        })
    return replies

Engagement Analytics Dashboard

The agent aggregates metrics across posts to identify content performance trends:

def compute_content_analytics(
    metrics: list[EngagementMetrics],
) -> dict:
    """Aggregate engagement metrics for content strategy insights."""
    if not metrics:
        return {"total_posts": 0}

    total_likes = sum(m.likes for m in metrics)
    total_comments = sum(m.comments for m in metrics)
    total_shares = sum(m.shares for m in metrics)
    total_impressions = sum(m.impressions for m in metrics)

    avg_engagement = (
        sum(m.engagement_rate for m in metrics) / len(metrics)
    ) if metrics else 0

    top_posts = sorted(metrics, key=lambda m: m.engagement_rate, reverse=True)[:5]

    return {
        "total_posts": len(metrics),
        "total_likes": total_likes,
        "total_comments": total_comments,
        "total_shares": total_shares,
        "total_impressions": total_impressions,
        "avg_engagement_rate": round(avg_engagement, 4),
        "top_post_ids": [p.post_id for p in top_posts],
    }

FAQ

How do I handle platform-specific content limits?

Each platform has different constraints: Twitter/X allows 280 characters, LinkedIn allows 3,000, and Instagram captions can be up to 2,200 characters. Add a max_length property to each platform client and validate content length before publishing. Use the LLM to adapt content for different platform limits while preserving the core message.

Should I auto-publish or require human approval?

Start with a review queue where the agent drafts posts and suggests schedules, but a human approves before publishing. Once you have validated the agent's output quality over 50 or more posts, enable auto-publishing for content types with consistent quality, like reshares and engagement replies, while keeping original content in the review queue.

How do I track which content topics perform best?

Tag each post with topic categories during creation. After collecting engagement data, group metrics by topic and compute average engagement rates per topic. The agent can then use this data to recommend more content on high-performing topics and suggest pivoting away from underperforming ones.

---

#SocialMedia #AIAgents #ContentScheduling #EngagementTracking #WorkflowAutomation #Python #AgenticAI #LearnAI #AIEngineering