FastAPI research agent service that orchestrates multi-step AI workflows with planning, tool use (Tavily, arXiv, Wikipedia), and Postgres state management
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
FastAPI research agent service that orchestrates multi-step AI workflows with planning, tool use (Tavily, arXiv, Wikipedia), and Postgres state management
triggers
["set up agentic research workflow service","create a research agent with planning and reflection","build multi-step AI research workflow with FastAPI","implement research agent with tool calling","set up reflective research agent with postgres","create research workflow with Tavily and arXiv","build agentic workflow with planning and execution","implement research report generation agent"]
The Agentic Research Workflow is a FastAPI-based service that implements a reflective, multi-step research agent system. It orchestrates planning, research, writing, and editing agents that work together to generate comprehensive research reports. The system uses Postgres for state management and supports tool-calling agents that can query Tavily (web search), arXiv (academic papers), and Wikipedia.
Key capabilities:
Multi-agent workflow orchestration (planner → research → writer → editor)
Tool-using agents with external API integration
Task state tracking and progress monitoring via REST API
import requests
import time
task_id = "550e8400-e29b-41d4-a716-446655440000"whileTrue:
response = requests.get(f"http://localhost:8000/task_progress/{task_id}")
data = response.json()
print(f"Status: {data['status']} - Current step: {data.get('current_step', 'N/A')}")
if data['status'] in ['completed', 'failed']:
break
time.sleep(2)
3. Get Final Report
GET /task_status/{task_id}
Response:
{"task_id":"550e8400-e29b-41d4-a716-446655440000","status":"completed","report":"# Research Report\n\n## Introduction...","created_at":"2025-01-15T10:30:00","updated_at":"2025-01-15T10:35:00"}
#!/bin/bashset -e
# Start Postgres
pg_ctlcluster $(pg_lsclusters -h | awk '{print $1}') main start
# Wait for Postgresuntil su -s /bin/bash postgres -c "psql -c 'SELECT 1'" > /dev/null 2>&1; dosleep 1
done# Create user and database
su -s /bin/bash postgres -c "psql -c \"CREATE ROLE app WITH LOGIN PASSWORD 'local';\""
su -s /bin/bash postgres -c "psql -c \"CREATE DATABASE appdb OWNER app;\""# Set DATABASE_URLexport DATABASE_URL="postgresql://app:local@127.0.0.1:5432/appdb"# Start FastAPIexec uvicorn main:app --host 0.0.0.0 --port 8000
Common Patterns
Threaded Task Execution
import threading
import uuid
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
defexecute_research_workflow(task_id: str, prompt: str, model: str):
"""
Background task that executes the full research workflow.
"""
db = SessionLocal()
try:
# Update status
task = db.query(Task).filter(Task.task_id == task_id).first()
task.status = "running"
db.commit()
# Execute plan
plan = planner_agent(prompt, model)
context = {"topic": prompt, "plan": plan}
for step in plan.get("steps", []):
result = executor_agent_step(step, context, model)
# Update progress in DB
task.progress = json.dumps({"current_step": step["name"], "result": result})
db.commit()
# Save final report
task.report = context.get("final_report", "No report generated")
task.status = "completed"
db.commit()
except Exception as e:
task.status = "failed"
task.report = f"Error: {str(e)}"
db.commit()
finally:
db.close()
@app.post("/generate_report")asyncdefgenerate_report(request: dict, background_tasks: BackgroundTasks):
"""
Kick off a research workflow in the background.
"""
task_id = str(uuid.uuid4())
prompt = request.get("prompt")
model = request.get("model", "openai:gpt-4o")
# Create task record
db = SessionLocal()
task = Task(task_id=task_id, prompt=prompt, model=model)
db.add(task)
db.commit()
db.close()
# Start background thread
thread = threading.Thread(
target=execute_research_workflow,
args=(task_id, prompt, model)
)
thread.start()
return {"task_id": task_id}
Progress Tracking
import json
@app.get("/task_progress/{task_id}")asyncdeftask_progress(task_id: str):
"""
Get detailed progress for a running task.
"""
db = SessionLocal()
task = db.query(Task).filter(Task.task_id == task_id).first()
db.close()
ifnot task:
return {"error": "Task not found"}
progress_data = {}
if task.progress:
progress_data = json.loads(task.progress)
return {
"task_id": task_id,
"status": task.status,
"current_step": progress_data.get("current_step"),
"progress": progress_data
}
Error Handling Pattern
defsafe_tool_call(tool_func, *args, **kwargs):
"""
Safely execute a tool call with error handling.
"""try:
return {
"success": True,
"data": tool_func(*args, **kwargs)
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": f"Network error: {str(e)}"
}
except Exception as e:
return {
"success": False,
"error": f"Tool error: {str(e)}"
}
# Usage in agent
result = safe_tool_call(tavily_search_tool, "quantum computing")
if result["success"]:
data = result["data"]
else:
print(f"Tool failed: {result['error']}")
Troubleshooting
Database Connection Issues
Problem:DATABASE_URL not set error
Solution:
# Verify environment variable is set
docker exec -it fpsvc bash -c 'echo $DATABASE_URL'# Should output: postgresql://app:local@127.0.0.1:5432/appdb# If empty, check entrypoint.sh exports it correctly
Problem: Tables not created
Solution:
# In main.py, ensure tables are created on startupfrom sqlalchemy import inspect
definit_db():
inspector = inspect(engine)
ifnot inspector.has_table("tasks"):
Base.metadata.create_all(bind=engine)
print("✅ Database tables created")
else:
print("✅ Database tables already exist")
# Call on startup
init_db()
API Key Issues
Problem: Tavily search fails with authentication error
Solution:
# Verify API key is loaded
docker exec -it fpsvc bash -c 'echo $TAVILY_API_KEY | head -c 20'# Ensure .env file is passed to docker run
docker run --env-file .env ...
# Or set explicitly
docker run -e TAVILY_API_KEY=your_key ...
Problem: OpenAI API errors
Solution:
# Add retry logic for API callsimport time
from openai import RateLimitError
defcall_llm_with_retry(client, **kwargs):
max_retries = 3for attempt inrange(max_retries):
try:
return client.chat.completions.create(**kwargs)
except RateLimitError:
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # Exponential backoffcontinueraise
# Stream large responses instead of loading all at oncefrom fastapi.responses import StreamingResponse
@app.get("/task_report_stream/{task_id}")asyncdefstream_report(task_id: str):
defgenerate():
db = SessionLocal()
task = db.query(Task).filter(Task.task_id == task_id).first()
db.close()
if task and task.report:
# Stream in chunks
chunk_size = 1024for i inrange(0, len(task.report), chunk_size):
yield task.report[i:i+chunk_size]
return StreamingResponse(generate(), media_type="text/markdown")
Docker Issues
Problem: Container exits immediately
Solution:
# Check logs
docker logs fpsvc
# Run interactively to debug
docker run --rm -it --entrypoint bash fastapi-postgres-service
# Inside container, manually run commands from entrypoint.sh
