How to Create an AI Content Publishing Agent
Build an agent that publishes content across platforms on schedule.
Jay Banlasan
The AI Systems Guy
The ai content publishing agent multi-platform I built handles publish content across platforms. I use this across client work where repetitive multi-step processes need to run without constant oversight.
The approach: adapt content for each platform format and publish simultaneously. One script, one run, results delivered.
What You Need
- Python 3.9+
- Anthropic API key
- Relevant API credentials for the tools your agent uses
Step 1: Define the Agent Tools
import anthropic
import json
import os
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic()
tools = [
{
"name": "adapt",
"description": "Primary tool for the agent's core function",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Input for the tool"}
},
"required": ["query"]
}
},
{
"name": "publish",
"description": "Secondary tool for processing or storing results",
"input_schema": {
"type": "object",
"properties": {
"data": {"type": "string", "description": "Data to process"}
},
"required": ["data"]
}
}
]
Step 2: Implement Tool Functions
def execute_tool(tool_name, tool_input):
if tool_name == "adapt":
return handle_adapt(tool_input)
elif tool_name == "publish":
return handle_publish(tool_input)
return "Unknown tool"
def handle_adapt(input_data):
# Your implementation here
query = input_data.get("query", "")
print(f"Running adapt: {query}")
return f"Results for: {query}"
def handle_publish(input_data):
data = input_data.get("data", "")
print(f"Processing: {data[:100]}")
return "Processed successfully"
Step 3: Build the Agent Loop
def run_agent(task, max_steps=10):
messages = [{"role": "user", "content": task}]
for step in range(max_steps):
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
system="You are an autonomous content publishing agent. Use the available tools to complete the task. Think step by step. Be thorough.",
tools=tools,
messages=messages
)
# Check if agent is done
if response.stop_reason == "end_turn":
final = next((b.text for b in response.content if b.type == "text"), "")
print(f"Agent completed in {step + 1} steps")
return final
# Process tool calls
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": str(result)
})
messages.append({"role": "user", "content": tool_results})
return "Max steps reached"
Step 4: Run and Log Results
import sqlite3
from datetime import datetime
def log_agent_run(task, result):
conn = sqlite3.connect("agent_runs.db")
conn.execute("""CREATE TABLE IF NOT EXISTS runs (
task TEXT, result TEXT, ran_at TEXT
)""")
conn.execute("INSERT INTO runs VALUES (?, ?, ?)",
(task, result[:5000], datetime.now().isoformat()))
conn.commit()
task = "Analyze our top competitors and create a summary report"
result = run_agent(task)
log_agent_run(task, result)
print(result)
What to Build Next
Add error recovery so the agent retries failed tool calls with adjusted parameters. Then add a cost tracker that monitors API token usage per agent run so you can optimize which model handles which steps.
Related Reading
- How to Create Social Proof Content with AI - creating and distributing content efficiently
- How to Create an AI-Powered Content Brief - creating and distributing content efficiently
- Prompt: Create a Content Distribution Plan - creating and distributing content efficiently
Want this system built for your business?
Get a free assessment. We will map every system your business needs and show you the ROI.
Get Your Free Assessment