How to Build an AI Task Planning Agent
Create an agent that breaks down goals into tasks and manages execution.
Jay Banlasan
The AI Systems Guy
The ai task planning agent project management I built handles break goals into executable tasks. I use this across client work where repetitive multi-step processes need to run without constant oversight.
The approach: take a high-level objective, decompose it into tasks with estimates, and generate a sprint plan. 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": "decompose",
"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": "estimate",
"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 == "decompose":
return handle_decompose(tool_input)
elif tool_name == "estimate":
return handle_estimate(tool_input)
return "Unknown tool"
def handle_decompose(input_data):
# Your implementation here
query = input_data.get("query", "")
print(f"Running decompose: {query}")
return f"Results for: {query}"
def handle_estimate(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 task planning 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
- Project Management with AI - how to pick the right automation approach for your business
- Building a Knowledge Management System - practical guidance for building AI-powered business systems
- Building an AI-Powered Calendar Management System - keeping schedules in sync without the back and forth
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