Introduction: What Is a Social Media Manager Agent?
A Social Media Manager Agent built with LangChain is an AI-powered autonomous system that can draft, schedule, analyze, and optimize social media content across multiple platforms. It combines the reasoning capabilities of large language models (LLMs) with concrete tools — API integrations, scheduling engines, analytics dashboards — to create a single intelligent assistant that handles the repetitive yet creative workload of a social media manager.
Unlike a static script or a simple chatbot, a LangChain agent uses a reasoning loop: it observes the current state (pending posts, engagement metrics, calendar events), decides which tool to invoke, executes that tool, and feeds the result back into its decision-making process. This means the agent can dynamically adapt — it might decide to rewrite a post that underperformed, reschedule content based on trending topics, or generate a completely new campaign from scratch based on a vague human directive like "boost our engagement next week."
Why a Social Media Manager Agent Matters
Social media management today involves a staggering amount of context-switching: jumping between X/Twitter, LinkedIn, Instagram, TikTok, and Facebook; tracking engagement metrics; responding to comments; maintaining brand voice; and staying on top of trending conversations. A well-built LangChain agent consolidates these workflows into a single conversational interface. Here's why this matters:
- Eliminates platform fatigue — The agent normalizes interactions across APIs so you issue one natural-language command instead of logging into five dashboards.
- Enforces brand consistency — Because the agent uses shared prompt templates and memory, it maintains tone, voice, and visual identity guidelines across every platform automatically.
- Closes the feedback loop — The agent doesn't just post; it reads analytics, compares performance against goals, and adjusts future content accordingly — something manual workflows rarely achieve in real time.
- Scales with your team — A single agent can serve multiple team members, each with different permission scopes, while maintaining a centralized content calendar and audit log.
- Reduces costly mistakes — Built-in guardrails, content policies, and approval chains prevent off-brand or legally risky posts from going live.
In short, the agent transforms social media from a chaotic, reactive chore into a strategic, data-driven operation — all while letting human creatives focus on high-level storytelling rather than copy-pasting between browser tabs.
Core Architecture Overview
Before diving into code, let's map the architecture. A LangChain social media agent typically consists of these layers:
- LLM Brain — The core reasoning engine (GPT-4o, Claude, Gemini, or a fine-tuned open model) that interprets instructions and decides which tool to use.
- Tool Belt — A collection of functions the agent can call: a post-drafting tool, a platform-publishing tool, an analytics fetcher, a calendar query tool, a trend-scraping tool, and a compliance checker.
- Memory — Short-term conversation memory plus a persistent vector store that remembers past campaigns, approved messaging, and performance history.
- Orchestrator — The LangChain AgentExecutor that runs the reasoning loop, handling tool errors, retries, and output parsing.
- Safety & Approval Layer — A human-in-the-loop checkpoint that intercepts high-risk actions (publishing, deleting, responding to sensitive comments) before execution.
We'll build each layer incrementally, starting with a minimal agent and progressively adding capabilities until we have a production-ready system.
Step-by-Step Implementation
1. Environment Setup and Dependencies
Create a new Python project and install the required packages. We'll use LangChain with OpenAI as the LLM provider, plus a few utilities for HTTP requests and scheduling.
# Create project directory
mkdir social-agent && cd social-agent
python -m venv .venv && source .venv/bin/activate
# Install core dependencies
pip install langchain langchain-openai langchain-community
pip install requests python-dotenv apscheduler
pip install pydantic rich # for structured output and pretty logging
Set up your .env file with API keys. You'll need an OpenAI key (or your preferred provider), and eventually keys for each social platform.
# .env
OPENAI_API_KEY=sk-your-key-here
TWITTER_BEARER_TOKEN=your-twitter-bearer-token
TWITTER_API_KEY=your-twitter-api-key
TWITTER_API_SECRET=your-twitter-api-secret
TWITTER_ACCESS_TOKEN=your-twitter-access-token
TWITTER_ACCESS_SECRET=your-twitter-access-secret
LINKEDIN_ACCESS_TOKEN=your-linkedin-token
META_ACCESS_TOKEN=your-meta-token
2. Defining the Tool Belt: Custom LangChain Tools
Tools are the agent's hands. Each tool is a Python function wrapped with a proper docstring and the @tool decorator. Let's build the core set.
First, a drafting tool that uses the LLM itself to generate post content according to platform-specific constraints and brand guidelines:
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from pydantic import BaseModel, Field
from typing import Optional, List
import json
# --- Structured input schemas improve reliability ---
class DraftPostInput(BaseModel):
platform: str = Field(description="Target platform: 'twitter', 'linkedin', 'instagram', or 'facebook'")
topic: str = Field(description="The core topic or message for the post")
tone: str = Field(default="professional", description="Desired tone: professional, casual, humorous, inspirational")
include_hashtags: bool = Field(default=True, description="Whether to include relevant hashtags")
max_length: Optional[int] = Field(default=None, description="Override default character limit if needed")
class DraftPostOutput(BaseModel):
platform: str
content: str
character_count: int
suggested_hashtags: List[str]
estimated_engagement_score: float
# --- The drafting tool ---
@tool(args_schema=DraftPostInput)
def draft_social_post(platform: str, topic: str, tone: str = "professional",
include_hashtags: bool = True, max_length: Optional[int] = None) -> str:
"""
Draft a social media post tailored to the specified platform.
Returns the drafted content along with metadata like character count and hashtag suggestions.
Use this whenever you need to create new social media content.
"""
# Platform-specific defaults
platform_config = {
"twitter": {"max_chars": 280, "hashtag_limit": 3, "style": "concise, punchy"},
"linkedin": {"max_chars": 3000, "hashtag_limit": 5, "style": "professional, narrative"},
"instagram": {"max_chars": 2200, "hashtag_limit": 30, "style": "visual-first, engaging"},
"facebook": {"max_chars": 63206, "hashtag_limit": 3, "style": "conversational, community-focused"}
}
config = platform_config.get(platform.lower(), platform_config["twitter"])
char_limit = max_length or config["max_chars"]
# Build a detailed prompt for the LLM
drafting_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = f"""You are a world-class social media copywriter. Draft a post for {platform} with these requirements:
- Topic: {topic}
- Tone: {tone}
- Platform style: {config['style']}
- Maximum characters: {char_limit} (strict — the post must fit)
- Include hashtags: {'Yes, up to ' + str(config['hashtag_limit']) + ' relevant hashtags' if include_hashtags else 'No hashtags'}
- Make it engaging and authentic. Avoid clichés.
- Return ONLY a valid JSON object with keys: "content" (the post text), "suggested_hashtags" (list), "estimated_engagement_score" (float 0-10).
DO NOT wrap the JSON in markdown code fences. Return raw JSON only."""
response = drafting_llm.invoke(prompt)
try:
parsed = json.loads(response.content.strip())
result = DraftPostOutput(
platform=platform,
content=parsed["content"],
character_count=len(parsed["content"]),
suggested_hashtags=parsed.get("suggested_hashtags", []),
estimated_engagement_score=parsed.get("estimated_engagement_score", 7.0)
)
return json.dumps(result.dict(), indent=2)
except json.JSONDecodeError:
# Fallback: return the raw content wrapped in a basic structure
fallback = DraftPostOutput(
platform=platform,
content=response.content.strip(),
character_count=len(response.content.strip()),
suggested_hashtags=[],
estimated_engagement_score=5.0
)
return json.dumps(fallback.dict(), indent=2)
Next, an analytics tool that fetches recent engagement metrics. In production, this would call real platform APIs; here we'll simulate with structured data for demonstration:
class FetchAnalyticsInput(BaseModel):
platform: str = Field(description="Platform to fetch analytics for")
days: int = Field(default=7, description="Number of days to look back")
metric_focus: str = Field(default="all", description="Specific metric: 'engagement', 'reach', 'clicks', 'all'")
@tool(args_schema=FetchAnalyticsInput)
def fetch_analytics(platform: str, days: int = 7, metric_focus: str = "all") -> str:
"""
Fetch recent social media analytics for the specified platform.
Returns engagement metrics, top-performing posts, and audience insights.
Use this to evaluate content performance and inform future strategy.
"""
# In production: call platform-specific APIs
# This simulation provides realistic structured data
analytics_data = {
"platform": platform,
"period_days": days,
"summary": {
"total_posts": 14,
"total_impressions": 87420,
"total_engagements": 3421,
"engagement_rate": 3.91,
"follower_change": +87,
"top_performing_hour": "10:00-11:00 UTC"
},
"top_posts": [
{
"content_snippet": "Excited to announce our new partnership with...",
"impressions": 12400,
"likes": 342,
"shares": 89,
"comments": 45,
"engagement_rate": 3.84
},
{
"content_snippet": "Behind the scenes at our product launch...",
"impressions": 9800,
"likes": 287,
"shares": 102,
"comments": 53,
"engagement_rate": 4.51
}
],
"audience_demographics": {
"top_regions": ["United States", "United Kingdom", "Canada"],
"age_groups": {"25-34": "38%", "35-44": "29%", "18-24": "18%"},
"gender_split": {"female": "52%", "male": "46%", "other": "2%"}
},
"recommendations": [
"Video content outperforms static images by 2.3x",
"Posts with questions in the caption see 41% more comments",
"Tuesday and Thursday mornings show highest organic reach"
]
}
return json.dumps(analytics_data, indent=2)
Now, a content calendar tool that reads and writes to a scheduled posts store. We'll use a simple JSON file for persistence:
import os
from datetime import datetime, timedelta
SCHEDULE_FILE = "content_calendar.json"
def _load_schedule() -> List[dict]:
if not os.path.exists(SCHEDULE_FILE):
return []
with open(SCHEDULE_FILE, "r") as f:
return json.load(f)
def _save_schedule(schedule: List[dict]) -> None:
with open(SCHEDULE_FILE, "w") as f:
json.dump(schedule, f, indent=2, default=str)
class SchedulePostInput(BaseModel):
platform: str = Field(description="Platform to schedule the post on")
content: str = Field(description="The exact post content to schedule")
scheduled_time: str = Field(description="ISO-formatted datetime string, e.g. '2025-03-15T14:30:00'")
tags: Optional[List[str]] = Field(default=None, description="Optional tags or campaign labels")
@tool(args_schema=SchedulePostInput)
def schedule_post(platform: str, content: str, scheduled_time: str, tags: Optional[List[str]] = None) -> str:
"""
Schedule a social media post for future publication.
Adds the post to the content calendar. Use this after drafting and approving content.
"""
schedule = _load_schedule()
entry = {
"id": f"post_{len(schedule) + 1:04d}",
"platform": platform,
"content": content,
"scheduled_time": scheduled_time,
"tags": tags or [],
"status": "scheduled",
"created_at": datetime.now().isoformat(),
"character_count": len(content)
}
schedule.append(entry)
_save_schedule(schedule)
return json.dumps({
"status": "success",
"message": f"Post scheduled for {platform} at {scheduled_time}",
"post_id": entry["id"],
"total_scheduled_posts": len(schedule)
}, indent=2)
class ViewCalendarInput(BaseModel):
platform_filter: Optional[str] = Field(default=None, description="Filter by platform, or omit for all")
days_ahead: int = Field(default=7, description="Number of days to look ahead")
@tool(args_schema=ViewCalendarInput)
def view_content_calendar(platform_filter: Optional[str] = None, days_ahead: int = 7) -> str:
"""
View the upcoming content calendar.
Shows all scheduled posts within the specified time window, optionally filtered by platform.
"""
schedule = _load_schedule()
now = datetime.now()
cutoff = now + timedelta(days=days_ahead)
upcoming = []
for entry in schedule:
try:
post_time = datetime.fromisoformat(entry["scheduled_time"])
if now <= post_time <= cutoff:
if platform_filter is None or entry["platform"].lower() == platform_filter.lower():
upcoming.append(entry)
except (KeyError, ValueError):
continue
upcoming.sort(key=lambda x: x.get("scheduled_time", ""))
return json.dumps({
"query_time": now.isoformat(),
"window": f"Now to {cutoff.isoformat()}",
"total_upcoming": len(upcoming),
"posts": upcoming
}, indent=2, default=str)
A compliance checker tool is critical for production. It screens posts against brand guidelines and legal policies before publishing:
class ComplianceCheckInput(BaseModel):
content: str = Field(description="The post content to check")
platform: str = Field(description="The target platform, which may have specific rules")
@tool(args_schema=ComplianceCheckInput)
def compliance_check(content: str, platform: str) -> str:
"""
Check a post against brand safety and compliance guidelines.
Returns a pass/fail verdict with specific issues flagged.
ALWAYS run this before scheduling or publishing any content.
"""
compliance_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
prompt = f"""You are a brand safety compliance auditor. Review this social media post for {platform}.
Content to review:
\"\"\"
{content}
\"\"\"
Check for these issues:
1. Offensive or discriminatory language
2. Factual claims without substantiation
3. Confidential or internal-only information
4. Competitor bashing
5. Regulatory compliance (financial advice, health claims, etc.)
6. Brand voice alignment (consistent with a professional, helpful, authentic tone)
7. Platform-specific rule violations ({platform} content policies)
Return a JSON object:
{{
"verdict": "PASS" or "FAIL",
"risk_level": "low", "medium", or "high",
"issues": [list specific issues found],
"suggested_fixes": [list actionable corrections],
"requires_human_review": true/false
}}
Return ONLY the JSON, no markdown fences."""
response = compliance_llm.invoke(prompt)
try:
parsed = json.loads(response.content.strip())
return json.dumps(parsed, indent=2)
except json.JSONDecodeError:
return json.dumps({
"verdict": "FAIL",
"risk_level": "medium",
"issues": ["Automated check inconclusive — raw response could not be parsed"],
"suggested_fixes": ["Manual review required"],
"requires_human_review": True
}, indent=2)
3. Assembling the Agent with LangChain
Now we wire everything together. The agent uses a ReAct-style prompt, a toolkit of our custom tools, and an AgentExecutor that manages the reasoning loop.
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from dotenv import load_dotenv
import logging
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("social-agent")
# --- Initialize the LLM ---
llm = ChatOpenAI(model="gpt-4o", temperature=0.2) # low temp for reliable tool selection
# --- Collect all tools ---
tools = [
draft_social_post,
fetch_analytics,
schedule_post,
view_content_calendar,
compliance_check
]
# --- Custom system prompt that defines the agent's role ---
system_prompt = """You are SocialBot, an expert AI social media manager agent. Your responsibilities:
1. Draft engaging, platform-optimized content when asked
2. Analyze engagement metrics and provide actionable insights
3. Schedule posts at optimal times based on analytics
4. Maintain and query the content calendar
5. Run compliance checks on ALL content before scheduling or publishing
6. Suggest content strategies based on performance data
CRITICAL RULES:
- NEVER schedule or publish a post without running compliance_check first
- When drafting, always consider the target platform's character limits and style
- When analyzing, provide specific, data-backed recommendations
- If a tool returns an error, diagnose the issue and try an alternative approach
- Be proactive: if analytics show declining engagement, suggest content experiments
- Maintain a helpful, strategic tone — you're a trusted marketing partner
Available tools: draft_social_post, fetch_analytics, schedule_post, view_content_calendar, compliance_check"""
# --- Build the prompt template ---
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
MessagesPlaceholder(variable_name="chat_history") # for memory
])
# --- Create the agent ---
agent = create_openai_tools_agent(llm, tools, prompt)
# --- Create memory for multi-turn conversations ---
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
max_token_limit=4000
)
# --- Build the executor with error handling ---
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
handle_parsing_errors=True,
max_iterations=8, # prevent infinite loops
max_execution_time=120, # seconds timeout
early_stopping_method="generate"
)
logger.info("Social Media Manager Agent initialized successfully with %d tools", len(tools))
4. Testing the Agent: Example Interactions
Let's run the agent through a realistic workflow. We'll simulate a marketing manager asking the agent to handle the morning social media routine.
# --- Example 1: Simple drafting ---
response = agent_executor.invoke({
"input": "Draft a LinkedIn post announcing our new AI-powered analytics feature.
Make it professional but exciting. Include relevant hashtags."
})
print(response["output"])
# --- Example 2: Full workflow with analytics + drafting + compliance + scheduling ---
response = agent_executor.invoke({
"input": """Here's what I need you to do step by step:
1. Fetch the last 7 days of Twitter analytics
2. Based on what performed well, draft a new Twitter post for tomorrow morning
3. Run compliance check on that draft
4. If it passes, schedule it for tomorrow at 9:00 AM UTC
5. Show me the updated calendar for the next 3 days"""
})
print(response["output"])
The agent will output its reasoning step by step (visible with verbose=True). You'll see tool calls, observations, and final synthesized responses. Here's what a typical interaction looks like internally:
# Agent trace (simplified):
# Thought: The user wants a full workflow. Let me start with analytics.
# Action: fetch_analytics(platform="twitter", days=7, metric_focus="all")
# Observation: {engagement_rate: 3.91, top_posts: [...], recommendations: [...]}
# Thought: Top posts show video content performs well. I'll draft a video-themed post.
# Action: draft_social_post(platform="twitter", topic="video content tips", tone="casual")
# Observation: {content: "🎥 Want 2.3x more engagement?...", character_count: 247, ...}
# Thought: Now I must run compliance check before scheduling.
# Action: compliance_check(content="🎥 Want 2.3x more engagement?...", platform="twitter")
# Observation: {verdict: "PASS", risk_level: "low", issues: [], requires_human_review: false}
# Thought: Compliance passed. Now I can schedule.
# Action: schedule_post(platform="twitter", content="🎥 Want 2.3x more engagement?...",
# scheduled_time="2025-03-16T09:00:00", tags=["video-tips", "engagement"])
# Observation: {status: "success", post_id: "post_0005", total_scheduled_posts: 5}
# Thought: All steps complete. Let me show the calendar.
# Action: view_content_calendar(days_ahead=3)
# Observation: {total_upcoming: 3, posts: [...]}
# Final Answer: Here's a summary of everything I did...
5. Adding Memory for Campaign Continuity
The ConversationBufferMemory we configured above stores the full conversation history. But for long-running campaigns, we need persistent memory that survives server restarts. Let's add a vector-backed memory store for campaign knowledge:
from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# --- Persistent vector memory for campaign history ---
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
collection_name="social_media_campaigns",
embedding_function=embeddings,
persist_directory="./campaign_memory"
)
# Initialize with some campaign context
vectorstore.add_texts(
texts=[
"Brand voice: Warm, knowledgeable, slightly witty. Never sarcastic or negative.",
"Q1 2025 campaign theme: 'Data-Driven Human Connections' — focus on AI + empathy.",
"Best posting times: Twitter 9-10 AM UTC, LinkedIn 8-9 AM and 4-5 PM UTC, Instagram 12-1 PM UTC.",
"Evergreen content pillars: Product tips, customer stories, industry thought leadership, behind-the-scenes.",
"Avoided topics: Politics, competitor comparisons, unverified statistics, speculative AI claims."
],
metadatas=[{"type": "brand_guideline"}] * 5,
ids=["guideline_1", "guideline_2", "guideline_3", "guideline_4", "guideline_5"]
)
# Create retriever memory
campaign_memory = VectorStoreRetrieverMemory(
retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
memory_key="campaign_context",
input_key="input"
)
# --- Enhanced agent with dual memory ---
# You can combine ConversationBufferMemory for dialogue + VectorStoreRetrieverMemory for knowledge
# For simplicity, we'll inject campaign context into the system prompt periodically
def enrich_prompt_with_memory(query: str) -> str:
"""Retrieve relevant campaign context and prepend to the system prompt."""
docs = vectorstore.similarity_search(query, k=3)
context = "\n".join([f"- {doc.page_content}" for doc in docs])
return f"{system_prompt}\n\n[RELEVANT CAMPAIGN CONTEXT]\n{context}\n[/CONTEXT]"
# Example: update system prompt before invocation
dynamic_prompt = enrich_prompt_with_memory("engagement strategy for next week")
# Then rebuild agent with dynamic_prompt as the system message
6. Integrating Real Social Media APIs
To graduate from simulation to production, you need actual API integrations. Here's a production-grade Twitter posting tool using the Twitter API v2:
import tweepy
from typing import Optional
class TwitterPostInput(BaseModel):
text: str = Field(description="Tweet text, max 280 characters")
reply_to_tweet_id: Optional[str] = Field(default=None, description="ID if this is a reply")
quote_tweet_id: Optional[str] = Field(default=None, description="ID if quote-tweeting")
@tool(args_schema=TwitterPostInput)
def publish_tweet(text: str, reply_to_tweet_id: Optional[str] = None,
quote_tweet_id: Optional[str] = None) -> str:
"""
ACTUALLY publish a tweet to Twitter/X via the API.
WARNING: This performs a real action. Only use after compliance_check passes.
"""
# Initialize Twitter client
client = tweepy.Client(
bearer_token=os.environ["TWITTER_BEARER_TOKEN"],
consumer_key=os.environ["TWITTER_API_KEY"],
consumer_secret=os.environ["TWITTER_API_SECRET"],
access_token=os.environ["TWITTER_ACCESS_TOKEN"],
access_token_secret=os.environ["TWITTER_ACCESS_SECRET"]
)
try:
if reply_to_tweet_id:
response = client.create_tweet(
text=text,
in_reply_to_tweet_id=reply_to_tweet_id
)
elif quote_tweet_id:
response = client.create_tweet(
text=text,
quote_tweet_id=quote_tweet_id
)
else:
response = client.create_tweet(text=text)
tweet_id = response.data["id"]
return json.dumps({
"status": "published",
"tweet_id": tweet_id,
"url": f"https://twitter.com/user/status/{tweet_id}",
"timestamp": datetime.now().isoformat()
}, indent=2)
except tweepy.TweepyException as e:
return json.dumps({
"status": "failed",
"error": str(e),
"suggestion": "Check character count, API rate limits, or authentication"
}, indent=2)
For LinkedIn, the integration follows a similar pattern using the LinkedIn Marketing API. Always wrap these in robust error handling:
# LinkedIn publishing tool (skeleton — requires LinkedIn API app setup)
@tool
def publish_linkedin(content: str, visibility: str = "PUBLIC",
share_media_category: str = "NONE") -> str:
"""
Publish a post to LinkedIn via the Marketing API.
Requires a valid LINKEDIN_ACCESS_TOKEN in environment.
"""
access_token = os.environ.get("LINKEDIN_ACCESS_TOKEN")
if not access_token:
return json.dumps({"status": "failed", "error": "LinkedIn token not configured"})
# LinkedIn share API endpoint
url = "https://api.linkedin.com/v2/ugcPosts"
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"X-Restli-Protocol-Version": "2.0.0"
}
payload = {
"author": f"urn:li:person:{{YOUR_PERSON_URN}}", # replace with actual URN
"lifecycleState": "PUBLISHED",
"specificContent": {
"com.linkedin.ugc.ShareContent": {
"shareCommentary": {"text": content},
"shareMediaCategory": share_media_category
}
},
"visibility": {
"com.linkedin.ugc.MemberNetworkVisibility": visibility
}
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=15)
response.raise_for_status()
return json.dumps({"status": "published", "response": response.json()}, indent=2)
except requests.RequestException as e:
return json.dumps({"status": "failed", "error": str(e)}, indent=2)
7. Building a Human-in-the-Loop Approval Chain
For high-stakes actions like publishing, you want human approval. LangChain supports interrupt-based workflows. Here's a pattern using a simple file-based approval queue:
import time
import hashlib
APPROVAL_FILE = "pending_approvals.json"
def submit_for_approval(action_type: str, details: dict) -> str:
"""Submit an action for human approval and return an approval token."""
approvals = _load_approvals()
approval_id = hashlib.sha256(f"{action_type}{json.dumps(details)}{time.time()}".encode()).hexdigest()[:12]
approvals.append({
"id": approval_id,
"action_type": action_type,
"details": details,
"submitted_at": datetime.now().isoformat(),
"status": "pending"
})
with open(APPROVAL_FILE, "w") as f:
json.dump(approvals, f, indent=2, default=str)
return json.dumps({
"approval_id": approval_id,
"message": f"Action '{action_type}' submitted for approval. Awaiting human review.",
"review_instructions": f"Set status to 'approved' or 'rejected' in {APPROVAL_FILE}"
}, indent=2)
def check_approval_status(approval_id: str) -> str:
"""Check if a submitted action has been approved."""
approvals = _load_approvals()
for approval in approvals:
if approval["id"] == approval_id:
return json.dumps(approval, indent=2, default=str)
return json.dumps({"error": "Approval ID not found"}, indent=2)
# --- Approval-aware publishing tool ---
@tool
def safe_publish_tweet(text: str) -> str:
"""
Submit a tweet for human approval before publishing.
This is the preferred publishing path for production use.
"""
# First, run compliance check automatically
compliance_result = json.loads(compliance_check.invoke({"content": text, "platform": "twitter"}))
if compliance_result.get("verdict") == "FAIL":
return json.dumps({
"status": "blocked",
"reason": "Compliance check failed",
"compliance_details": compliance_result
}, indent=2)
# Submit for human approval
approval_response = submit_for_approval("publish_tweet", {
"content": text,
"character_count": len(text),
"compliance_verdict": compliance_result.get("verdict"),
"risk_level": compliance_result.get("risk_level")
})
return approval_response
8. Scheduling and Automation with APScheduler
For autonomous operation, the agent needs a scheduling backbone that can execute publishing at predetermined times. APScheduler integrates cleanly:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import threading
class SocialMediaScheduler:
"""Background scheduler that publishes approved posts at their scheduled times."""
def __init__(self, agent_executor: AgentExecutor):
self.agent = agent_executor
self.scheduler = BackgroundScheduler()
self.scheduler.start()
logger.info("Background scheduler started")
def publish_scheduled_post(self, post_id: str) -> None:
"""Called by the scheduler at the appointed time."""
schedule =