What is a Social Media Manager Agent?
A Social Media Manager Agent is an autonomous AI-powered system that handles the end-to-end workflow of managing social media presence. Instead of manually logging into multiple platforms, drafting posts, tracking engagement, and responding to comments, you delegate these tasks to an intelligent agent that understands context, makes decisions, and executes actions across platforms.
The agent operates as a stateful orchestrator — it receives a high-level instruction like "Create and schedule a week's worth of Twitter posts about our product launch, then monitor engagement and respond to any negative comments" and breaks it down into discrete steps: content research, draft generation, platform-specific formatting, scheduling, continuous monitoring, sentiment analysis, and automated response drafting. Each step feeds into the next, with the agent maintaining memory of what it has done and adapting based on intermediate results.
Why LangGraph for Agentic Workflows?
LangGraph is a framework from the LangChain ecosystem designed specifically for building stateful, multi-actor agent applications. Unlike simple linear chains or single-turn tool-calling loops, LangGraph gives you fine-grained control over the agent's execution graph — nodes, edges, conditional branching, and persistent state management.
For a social media manager, this matters because:
- Multi-step persistence — The agent must remember scheduled posts across sessions and not lose state if a monitoring loop runs for days
- Conditional routing — A single user request may require the agent to choose between content creation, scheduling, analytics retrieval, or engagement response based on intent classification
- Human-in-the-loop — Before posting sensitive content, the workflow can pause and wait for human approval, then resume exactly where it left off
- Streaming and observability — You can inspect every intermediate state, tool call, and decision the agent makes, which is critical for debugging social media automation
- Parallel execution — The agent can simultaneously post to Twitter, LinkedIn, and Facebook using fan-out node patterns
LangGraph models your agent as a directed graph where each node is a processing step (an LLM call, a tool execution, or a custom function) and edges define how state flows between steps. The framework handles state serialization, checkpointing, and graph execution with built-in support for streaming updates.
Prerequisites and Environment Setup
Before building the agent, install the required packages. You'll need LangGraph, LangChain, an LLM provider (OpenAI is used here), and utility libraries for date handling and API requests.
# Create a virtual environment
python -m venv smm-agent-env
source smm-agent-env/bin/activate # On Windows: smm-agent-env\Scripts\activate
# Install core dependencies
pip install langgraph langchain langchain-openai openai
pip install python-dotenv requests python-dateutil pydantic
pip install httpx # For async API calls to social platforms
# For persistent checkpointing (optional but recommended)
pip install langgraph-checkpoint-sqliteSet up your OpenAI API key in a .env file:
# .env
OPENAI_API_KEY=sk-your-key-here
TWITTER_API_KEY=your-twitter-api-key
TWITTER_API_SECRET=your-twitter-api-secret
LINKEDIN_ACCESS_TOKEN=your-linkedin-tokenStep 1: Defining the Agent State
The state is the single source of truth that flows through every node in the graph. For a social media manager, the state must carry the user's original request, the agent's internal reasoning, generated content drafts, scheduling metadata, engagement data fetched from platforms, and a running log of actions taken.
Use a TypedDict to define the state schema so LangGraph can enforce type safety and serialize state between nodes:
from typing import TypedDict, Annotated, List, Dict, Optional, Literal
from datetime import datetime
import operator
class SocialMediaState(TypedDict):
# User input and agent reasoning
user_request: str
intent: Literal["create_content", "schedule_post", "monitor_engagement",
"respond_comment", "analytics_report", "general_chat"]
reasoning: str
# Content generation outputs
generated_posts: List[Dict[str, str]] # [{"platform": "twitter", "content": "...", "media_url": ""}]
content_approved: bool
# Scheduling metadata
schedule_times: List[Dict[str, str]] # [{"platform": "twitter", "datetime_iso": "..."}]
scheduled_confirmation: List[Dict[str, str]]
# Engagement and monitoring
platform_metrics: Dict[str, Dict] # {"twitter": {"likes": 10, "comments": [...]}}
pending_comments: List[Dict[str, str]]
response_drafts: List[Dict[str, str]]
# Execution log
action_log: Annotated[List[str], operator.add] # Appended across nodes using operator.add reducer
error_count: int
max_retries: int
# Human-in-the-loop flag
requires_approval: boolThe Annotated[List[str], operator.add] pattern tells LangGraph to concatenate new log entries rather than overwrite them — this is essential for maintaining an audit trail across the entire agent run.
Step 2: Building the Core Tools
Tools are functions the agent can invoke to interact with external services. Each tool should be self-contained, handle its own errors gracefully, and return structured results that the agent can incorporate into its reasoning.
Create a tools module with platform-specific operations:
import json
import httpx
from datetime import datetime, timedelta
from typing import List, Dict
import os
# ---------- Twitter (X) API Tools ----------
async def twitter_create_post(content: str, media_url: str = None) -> Dict:
"""Create a new tweet. Returns the tweet ID and timestamp."""
# In production, use the Twitter API v2 via OAuth 2.0
# Mock implementation for development
tweet_id = f"tweet_{int(datetime.now().timestamp())}"
return {
"platform": "twitter",
"post_id": tweet_id,
"content": content[:280], # Twitter character limit
"media_attached": media_url is not None,
"created_at": datetime.now().isoformat(),
"status": "posted"
}
async def twitter_get_engagement(post_id: str = None, days: int = 7) -> Dict:
"""Fetch engagement metrics for tweets. If post_id is None, returns account-level metrics."""
# Mock response — replace with real Twitter API calls
return {
"platform": "twitter",
"metrics": {
"impressions": 12450,
"likes": 342,
"retweets": 87,
"replies": 56,
"quote_tweets": 12,
"engagement_rate": 0.042
},
"top_comments": [
{"id": "c1", "text": "Love this product!", "sentiment": "positive"},
{"id": "c2", "text": "When will this ship internationally?", "sentiment": "neutral"},
{"id": "c3", "text": "Disappointed with the pricing", "sentiment": "negative"}
],
"fetched_at": datetime.now().isoformat()
}
async def twitter_respond_to_comment(comment_id: str, response_text: str) -> Dict:
"""Reply to a specific tweet/comment."""
return {
"platform": "twitter",
"comment_id": comment_id,
"response": response_text[:280],
"responded_at": datetime.now().isoformat(),
"status": "replied"
}
# ---------- LinkedIn API Tools ----------
async def linkedin_create_post(content: str, media_url: str = None) -> Dict:
"""Create a LinkedIn post using the LinkedIn Marketing API."""
post_id = f"li_post_{int(datetime.now().timestamp())}"
return {
"platform": "linkedin",
"post_id": post_id,
"content": content[:3000],
"media_attached": media_url is not None,
"created_at": datetime.now().isoformat(),
"status": "posted"
}
async def linkedin_get_engagement(post_id: str = None, days: int = 7) -> Dict:
"""Fetch LinkedIn post engagement metrics."""
return {
"platform": "linkedin",
"metrics": {
"impressions": 8900,
"likes": 215,
"comments": 34,
"shares": 18,
"click_through_rate": 0.031
},
"top_comments": [
{"id": "lc1", "text": "Great insights, thanks for sharing!", "sentiment": "positive"},
{"id": "lc2", "text": "We implemented this and saw 3x ROI", "sentiment": "positive"}
],
"fetched_at": datetime.now().isoformat()
}
# ---------- Unified Tool Registry ----------
TOOL_REGISTRY = {
"twitter_create_post": twitter_create_post,
"twitter_get_engagement": twitter_get_engagement,
"twitter_respond_to_comment": twitter_respond_to_comment,
"linkedin_create_post": linkedin_create_post,
"linkedin_get_engagement": linkedin_get_engagement,
}Each tool returns a standardized dictionary so the agent can parse results consistently. In a production system, replace the mock implementations with real API clients using httpx or the platform-specific SDKs, and add proper authentication handling, rate limit awareness, and exponential backoff.
Step 3: Constructing the LangGraph Workflow
Now build the graph. A LangGraph workflow consists of nodes (Python functions that read and modify state) and edges (connections that define the flow). The graph is compiled into an executable that handles state propagation automatically.
Start by defining the node functions that form the backbone of the agent:
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
import asyncio
# Initialize the LLM (used for reasoning, routing, and content generation)
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
# ---------- Node: Intent Classifier ----------
async def classify_intent(state: SocialMediaState) -> SocialMediaState:
"""Analyze the user request and determine the primary intent."""
system_prompt = """You are an intent classifier for a social media manager agent.
Classify the user's request into EXACTLY ONE of these categories:
- create_content: User wants to draft new social media posts
- schedule_post: User wants to schedule posts for future publication
- monitor_engagement: User wants to check metrics, likes, comments
- respond_comment: User wants to reply to specific comments
- analytics_report: User wants a comprehensive analytics summary
- general_chat: Casual conversation or clarification questions
Respond ONLY with the intent label, nothing else."""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=state["user_request"])
]
response = await llm.ainvoke(messages)
intent = response.content.strip().lower()
state["intent"] = intent
state["reasoning"] = f"Classified intent as: {intent}"
state["action_log"] = [f"[{datetime.now().isoformat()}] Intent classified: {intent}"]
return state
# ---------- Node: Content Generator ----------
async def generate_content(state: SocialMediaState) -> SocialMediaState:
"""Generate platform-optimized social media posts based on user request."""
system_prompt = """You are a professional social media content creator.
Given the user's request, generate engaging, platform-appropriate posts.
Rules:
- Twitter/X: Max 280 characters, use hashtags sparingly (2-3), conversational tone
- LinkedIn: Professional tone, 1200-3000 characters, use line breaks for readability
- Include relevant emojis where appropriate
- Each post must be self-contained and value-driven
Output valid JSON as a list of objects with 'platform' and 'content' keys.
Example: [{"platform": "twitter", "content": "..."}, {"platform": "linkedin", "content": "..."}]"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"Generate social media posts for: {state['user_request']}")
]
response = await llm.ainvoke(messages)
# Parse the JSON response
import re
json_match = re.search(r'\[.*\]', response.content, re.DOTALL)
if json_match:
posts = json.loads(json_match.group())
state["generated_posts"] = posts
else:
# Fallback: treat as raw text
state["generated_posts"] = [{"platform": "twitter", "content": response.content[:280]}]
state["requires_approval"] = True # Flag for human review before posting
state["action_log"] = [f"[{datetime.now().isoformat()}] Generated {len(state['generated_posts'])} posts"]
return state
# ---------- Node: Engagement Monitor ----------
async def monitor_engagement(state: SocialMediaState) -> SocialMediaState:
"""Fetch engagement data from connected social platforms."""
platforms_to_check = ["twitter", "linkedin"]
all_metrics = {}
all_comments = []
for platform in platforms_to_check:
tool_key = f"{platform}_get_engagement"
if tool_key in TOOL_REGISTRY:
result = await TOOL_REGISTRY[tool_key]()
all_metrics[platform] = result.get("metrics", {})
all_comments.extend(result.get("top_comments", []))
state["platform_metrics"] = all_metrics
state["pending_comments"] = all_comments
state["action_log"] = [f"[{datetime.now().isoformat()}] Fetched engagement from {len(platforms_to_check)} platforms"]
return state
# ---------- Node: Response Drafter ----------
async def draft_responses(state: SocialMediaState) -> SocialMediaState:
"""Draft appropriate responses to pending comments based on sentiment."""
if not state.get("pending_comments"):
state["response_drafts"] = []
return state
comments_json = json.dumps(state["pending_comments"], indent=2)
system_prompt = """You are a brand voice specialist. Draft personalized responses to social media comments.
Guidelines:
- Positive comments: Express gratitude, add value with a tip or resource
- Neutral/Questions: Answer helpfully, include a link if relevant
- Negative comments: Acknowledge the concern, remain professional, offer to resolve via DM
Output JSON array with 'comment_id' and 'response' for each comment that needs a reply."""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"Here are pending comments:\n{comments_json}\n\nDraft responses for each.")
]
response = await llm.ainvoke(messages)
json_match = re.search(r'\[.*\]', response.content, re.DOTALL)
if json_match:
drafts = json.loads(json_match.group())
state["response_drafts"] = drafts
else:
state["response_drafts"] = []
state["action_log"] = [f"[{datetime.now().isoformat()}] Drafted {len(state['response_drafts'])} responses"]
return state
# ---------- Node: Analytics Reporter ----------
async def generate_analytics(state: SocialMediaState) -> SocialMediaState:
"""Compile a comprehensive analytics report from collected metrics."""
metrics_json = json.dumps(state.get("platform_metrics", {}), indent=2)
system_prompt = """You are a social media analytics expert. Create a concise, actionable report.
Include: key metrics per platform, trends compared to previous period, top-performing content patterns,
and 3 specific recommendations for improving engagement."""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"Platform metrics data:\n{metrics_json}\n\nGenerate an analytics report.")
]
response = await llm.ainvoke(messages)
# Store the report in state — can be formatted for email/Slack later
state["action_log"] = [
f"[{datetime.now().isoformat()}] Analytics report generated: {response.content[:200]}..."
]
return stateEach node is a pure async function that takes the full state, transforms it, and returns it. LangGraph automatically manages the state object between nodes — you don't need to manually pass it around.
Step 4: Adding Conditional Routing Logic
The power of LangGraph comes from conditional edges — the ability to dynamically choose which node executes next based on the current state. For a social media manager, routing depends on the classified intent.
Define a router function and the graph structure with conditional branching:
# ---------- Router Function ----------
def route_by_intent(state: SocialMediaState) -> str:
"""Determine the next node based on classified intent."""
intent = state.get("intent", "general_chat")
routing_map = {
"create_content": "generate_content",
"schedule_post": "schedule_posts",
"monitor_engagement": "monitor_engagement",
"respond_comment": "draft_responses",
"analytics_report": "generate_analytics",
"general_chat": "handle_chat"
}
return routing_map.get(intent, "handle_chat")
# ---------- Additional Nodes ----------
async def schedule_posts(state: SocialMediaState) -> SocialMediaState:
"""Schedule generated posts for future publication."""
posts = state.get("generated_posts", [])
if not posts:
# If no posts generated yet, route back to content generation
state["intent"] = "create_content"
return state
# Schedule posts at optimal times (simplified — use actual timezone-aware scheduling in production)
schedule_map = {
"twitter": (datetime.now() + timedelta(hours=2)).isoformat(),
"linkedin": (datetime.now() + timedelta(hours=4)).isoformat(),
}
scheduled = []
for post in posts:
platform = post.get("platform", "twitter")
scheduled.append({
"platform": platform,
"content": post.get("content", "")[:280],
"scheduled_for": schedule_map.get(platform, datetime.now().isoformat()),
"status": "scheduled"
})
state["scheduled_confirmation"] = scheduled
state["action_log"] = [f"[{datetime.now().isoformat()}] Scheduled {len(scheduled)} posts"]
return state
async def handle_chat(state: SocialMediaState) -> SocialMediaState:
"""Handle general conversation or clarification requests."""
system_prompt = """You are a helpful social media management assistant.
Answer the user's question clearly and concisely. If they need help with specific tasks,
remind them you can create content, schedule posts, monitor engagement, respond to comments,
or generate analytics reports."""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=state["user_request"])
]
response = await llm.ainvoke(messages)
state["reasoning"] = response.content
state["action_log"] = [f"[{datetime.now().isoformat()}] Chat response provided"]
return state
async def human_approval_gate(state: SocialMediaState) -> SocialMediaState:
"""Node that pauses execution for human approval (human-in-the-loop pattern)."""
# In production, this would emit an event and wait for external input
# For now, we simulate approval with a flag
state["content_approved"] = True # Auto-approve in dev; set to False and add interrupt in production
state["action_log"] = [f"[{datetime.now().isoformat()}] Content approval gate passed"]
return state
# ---------- Build the Graph ----------
def build_social_media_agent() -> StateGraph:
"""Construct the full LangGraph workflow for the social media manager."""
workflow = StateGraph(SocialMediaState)
# Add all nodes to the graph
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("generate_content", generate_content)
workflow.add_node("schedule_posts", schedule_posts)
workflow.add_node("monitor_engagement", monitor_engagement)
workflow.add_node("draft_responses", draft_responses)
workflow.add_node("generate_analytics", generate_analytics)
workflow.add_node("handle_chat", handle_chat)
workflow.add_node("human_approval_gate", human_approval_gate)
# Set the entry point — where execution begins
workflow.set_entry_point("classify_intent")
# Add conditional edges from intent classification
workflow.add_conditional_edges(
"classify_intent",
route_by_intent,
{
"generate_content": "generate_content",
"schedule_posts": "schedule_posts",
"monitor_engagement": "monitor_engagement",
"draft_responses": "draft_responses",
"generate_analytics": "generate_analytics",
"handle_chat": "handle_chat"
}
)
# After content generation, route through approval gate, then optionally to scheduling
workflow.add_edge("generate_content", "human_approval_gate")
# Conditional edge after approval: if posts exist, offer scheduling; otherwise end
workflow.add_conditional_edges(
"human_approval_gate",
lambda state: "schedule_posts" if state.get("generated_posts") and state.get("content_approved") else END,
{
"schedule_posts": "schedule_posts",
END: END
}
)
# After scheduling, monitoring, responses, analytics, and chat — all end the run
workflow.add_edge("schedule_posts", END)
workflow.add_edge("monitor_engagement", END)
workflow.add_edge("draft_responses", END)
workflow.add_edge("generate_analytics", END)
workflow.add_edge("handle_chat", END)
return workflow
# Compile the graph
social_agent_graph = build_social_media_agent()
compiled_agent = social_agent_graph.compile()The conditional edge at classify_intent acts as the agent's decision-making hub. Based on the LLM-classified intent, execution branches to entirely different workflows — content creation flows through approval to scheduling, while monitoring jumps straight to data fetching. This topology mirrors how a human social media manager would mentally route incoming requests.
Step 5: Integrating Real Social Media API Operations
While the mock tools above are sufficient for development and testing, production integration requires proper API clients. Here's how to build a robust Twitter API v2 integration using OAuth 2.0:
import httpx
import os
from typing import Optional, Dict, List
class TwitterAPIClient:
"""Production-ready Twitter API v2 client with rate limiting and retry logic."""
def __init__(self):
self.api_key = os.getenv("TWITTER_API_KEY")
self.api_secret = os.getenv("TWITTER_API_SECRET")
self.access_token = os.getenv("TWITTER_ACCESS_TOKEN")
self.base_url = "https://api.twitter.com/2"
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json"
},
timeout=30.0
)
self.rate_limit_remaining = 300 # Default Twitter rate limit
self.rate_limit_reset = 0
async def _handle_rate_limit(self, response: httpx.Response):
"""Extract rate limit info and wait if necessary."""
if "x-rate-limit-remaining" in response.headers:
self.rate_limit_remaining = int(response.headers["x-rate-limit-remaining"])
if "x-rate-limit-reset" in response.headers:
reset_time = int(response.headers["x-rate-limit-reset"])
import time
wait_seconds = max(0, reset_time - int(time.time()))
if wait_seconds > 0 and self.rate_limit_remaining == 0:
import asyncio
await asyncio.sleep(wait_seconds + 1)
async def create_tweet(self, text: str, reply_settings: str = "everyone") -> Dict:
"""Post a tweet using the Twitter API v2."""
payload = {
"text": text[:280],
"reply_settings": reply_settings
}
# Retry with exponential backoff
max_retries = 3
for attempt in range(max_retries):
try:
response = await self.client.post("/tweets", json=payload)
await self._handle_rate_limit(response)
if response.status_code == 201:
data = response.json()
return {
"platform": "twitter",
"post_id": data["data"]["id"],
"content": text[:280],
"created_at": data.get("data", {}).get("created_at", ""),
"status": "posted"
}
elif response.status_code == 429:
# Rate limited — wait and retry
import asyncio
await asyncio.sleep(2 ** attempt * 5)
continue
else:
return {"platform": "twitter", "status": "failed", "error": response.text}
except Exception as e:
if attempt == max_retries - 1:
return {"platform": "twitter", "status": "failed", "error": str(e)}
import asyncio
await asyncio.sleep(2 ** attempt)
return {"platform": "twitter", "status": "failed", "error": "Max retries exceeded"}
async def get_tweet_metrics(self, tweet_id: str) -> Dict:
"""Fetch metrics for a specific tweet."""
response = await self.client.get(
f"/tweets/{tweet_id}",
params={"tweet.fields": "public_metrics,created_at"}
)
await self._handle_rate_limit(response)
if response.status_code == 200:
data = response.json()
metrics = data.get("data", {}).get("public_metrics", {})
return {
"platform": "twitter",
"tweet_id": tweet_id,
"metrics": metrics,
"fetched_at": datetime.now().isoformat()
}
return {"platform": "twitter", "error": response.text}
async def reply_to_tweet(self, tweet_id: str, reply_text: str) -> Dict:
"""Reply to an existing tweet."""
payload = {
"text": reply_text[:280],
"reply": {"in_reply_to_tweet_id": tweet_id}
}
response = await self.client.post("/tweets", json=payload)
await self._handle_rate_limit(response)
if response.status_code == 201:
data = response.json()
return {
"platform": "twitter",
"reply_to": tweet_id,
"reply_id": data["data"]["id"],
"response_text": reply_text[:280],
"status": "replied"
}
return {"platform": "twitter", "status": "failed", "error": response.text}This client handles rate limiting gracefully by reading Twitter's rate-limit headers and automatically sleeping when limits are exhausted. The exponential backoff with jitter prevents thundering-herd retries. Wrap this client in your LangGraph tool functions to bridge the gap between the agent's decision-making and real platform operations.
Step 6: Content Generation with Platform-Specific Optimization
One of the most valuable capabilities of a social media manager agent is generating content that's natively optimized for each platform's audience, tone, and constraints. The agent should understand that a LinkedIn post requires a different voice than a tweet, and should format accordingly.
Enhance the content generation node with platform-specific guardrails:
# ---------- Enhanced Content Generator with Platform Guardrails ----------
PLATFORM_CONSTRAINTS = {
"twitter": {
"max_length": 280,
"hashtag_limit": 3,
"tone": "conversational, witty, concise",
"best_practices": "Use line breaks for emphasis, include one strong CTA, avoid thread fatigue"
},
"linkedin": {
"max_length": 3000,
"hashtag_limit": 5,
"tone": "professional, insightful, value-driven",
"best_practices": "Start with a hook line, use whitespace formatting, end with a question to drive comments"
},
"facebook": {
"max_length": 63206,
"hashtag_limit": 2,
"tone": "warm, community-oriented, story-driven",
"best_practices": "Use native video or images, keep text concise, engage with polls"
},
"instagram": {
"max_length": 2200,
"hashtag_limit": 30,
"tone": "visual-first, aspirational, engaging",
"best_practices": "Lead with the visual description, use emojis naturally, place hashtags in first comment"
}
}
async def generate_platform_content(
user_request: str,
platforms: List[str],
brand_voice: str = "professional"
) -> List[Dict[str, str]]:
"""Generate content optimized for each specified platform."""
platform_details = []
for p in platforms:
if p in PLATFORM_CONSTRAINTS:
details = PLATFORM_CONSTRAINTS[p]
platform_details.append(
f"- {p}: max {details['max_length']} chars, "
f"tone: {details['tone']}, "
f