What is an AI Agent for Fitness Coaching?
An AI agent for fitness coaching is an autonomous software system that uses large language models (LLMs) to handle routine client management tasks ā think of it as a tireless virtual assistant that operates 24/7. Unlike a static chatbot that merely answers questions, an agent actively reasons about client data, decides which actions to take, executes them, and learns from the results. In the fitness context, this means the agent can send personalized check-in messages, analyze workout logs, flag clients who are falling behind, suggest schedule adjustments, and even draft motivational content ā all without the coach lifting a finger.
At its core, the agent follows a sense ā reason ā act loop. It senses incoming data (a client's completed workout, a missed session, a nutrition log update), reasons about what that data means using an LLM, and then acts by updating a database, sending an email, or scheduling a follow-up. The agent is equipped with a set of tools ā functions it can call ā such as get_client_profile(), send_email(), query_workout_history(), or create_calendar_event(). The LLM decides which tool to invoke and with what parameters, based on the current context.
Here is a concrete example: a client named Sarah misses her Monday morning workout. The agent detects the missed check-in, retrieves Sarah's history (she's missed two workouts this month), reasons that a gentle re-engagement message is appropriate, and autonomously sends her a personalized SMS: "Hey Sarah! I noticed you've had a couple of tough weeks ā completely normal. How about we adjust your plan for this week to make it more manageable? I'm here when you're ready." All of this happens in seconds, with no manual intervention from the coach.
Why AI-Powered Client Management Matters
Fitness coaches typically juggle 20 to 40 clients simultaneously. The administrative overhead of tracking check-ins, analyzing progress photos, adjusting schedules, and sending follow-ups consumes hours each day ā time that could be spent on high-value coaching activities like program design and one-on-one consultations. AI agents compress this administrative burden dramatically.
The key benefits break down into four categories:
- Scalability without quality loss ā A coach using an AI agent can effectively manage 3x to 5x more clients because the agent handles the repetitive touchpoints. Each client still receives personalized attention because the agent tailors messages based on individual history.
- 24/7 responsiveness ā Clients often work out at odd hours. An agent processes check-ins, answers common questions, and provides feedback instantly, regardless of when the activity occurs.
- Data-driven early interventions ā Agents continuously monitor adherence metrics (workout completion rate, log consistency, progress velocity) and flag at-risk clients days or weeks before a human coach would notice. This reduces churn by 15-30% based on early adopter reports.
- Consistent brand voice at scale ā The agent enforces the coach's communication style, tone, and follow-up cadence across every client interaction, eliminating the inconsistency that creeps in when a coach is rushing through a busy day.
From a technical perspective, this is not merely a convenience ā it is an architectural shift from reactive coaching (coach responds when client reaches out) to proactive coaching (system anticipates needs and acts first). The agent becomes the operational backbone of the coaching business.
Building Your First Fitness Coach AI Agent
Setting Up the Environment
We will build a production-ready agent using Python, FastAPI for the API layer, LangChain for the agent orchestration, PostgreSQL for persistence, and Twilio/SendGrid for messaging. Start by installing the dependencies:
# requirements.txt
fastapi==0.104.1
uvicorn==0.24.0
langchain==0.1.0
langchain-openai==0.0.5
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
pydantic==2.5.0
sendgrid==6.11.0
twilio==9.0.0
python-dotenv==1.0.0
alembic==1.13.0
redis==5.0.1
celery==5.3.6Set up your environment variables in a .env file:
# .env
OPENAI_API_KEY=sk-your-key-here
DATABASE_URL=postgresql://user:password@localhost:5432/fitness_agent
SENDGRID_API_KEY=SG.your-sendgrid-key
TWILIO_ACCOUNT_SID=your-account-sid
TWILIO_AUTH_TOKEN=your-auth-token
TWILIO_FROM_NUMBER=+15551234567
REDIS_URL=redis://localhost:6379/0Core Agent Architecture
The agent follows a tool-calling pattern where the LLM is given a set of function signatures and decides which to invoke. Here is the foundational agent class:
# agent/core.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from typing import Dict, Any
import json
class FitnessCoachAgent:
"""Autonomous agent that manages fitness coaching client operations."""
def __init__(self, tools: list, system_prompt: str):
self.llm = ChatOpenAI(
model="gpt-4-turbo-preview",
temperature=0.7,
max_tokens=1000
)
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("system", "You are an AI fitness coach assistant. "
"Your role is to monitor clients, send appropriate messages, "
"flag concerns, and suggest schedule adjustments. "
"Always prioritize client safety and well-being. "
"Never recommend drastic changes without coach approval. "
"Current timestamp: {current_time}"),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(self.llm, tools, prompt)
self.executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=8,
handle_parsing_errors=True,
return_intermediate_steps=True
)
async def run(self, input_text: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
"""Execute the agent with the given input and optional context."""
result = await self.executor.ainvoke({
"input": input_text,
"current_time": str(context.get("current_time", "")),
"chat_history": context.get("chat_history", [])
})
return {
"output": result["output"],
"intermediate_steps": result.get("intermediate_steps", []),
"tool_calls_made": [
step for step in result.get("intermediate_steps", [])
if hasattr(step, 'tool')
]
}
def run_sync(self, input_text: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
"""Synchronous wrapper for environments without async support."""
import asyncio
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(self.run(input_text, context))
finally:
loop.close()The agent executor manages the iterative loop: the LLM receives the prompt, decides on a tool call, the executor invokes that tool, feeds the result back to the LLM, and repeats until the LLM emits a final text response. The max_iterations parameter prevents infinite loops.
Client Data Management
Before the agent can act intelligently, it needs structured client data. Here are the SQLAlchemy models and a repository layer:
# models/client.py
from sqlalchemy import Column, String, Integer, Float, DateTime, Boolean, Enum, ForeignKey, Text
from sqlalchemy.orm import relationship, declarative_base
from sqlalchemy.dialects.postgresql import UUID, JSONB
import enum
import uuid
from datetime import datetime
Base = declarative_base()
class FitnessGoal(str, enum.Enum):
WEIGHT_LOSS = "weight_loss"
MUSCLE_GAIN = "muscle_gain"
ENDURANCE = "endurance"
STRENGTH = "strength"
GENERAL_FITNESS = "general_fitness"
COMPETITION_PREP = "competition_prep"
class ClientStatus(str, enum.Enum):
ACTIVE = "active"
AT_RISK = "at_risk"
PAUSED = "paused"
INACTIVE = "inactive"
class Client(Base):
__tablename__ = "clients"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
first_name = Column(String(100), nullable=False)
last_name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False)
phone = Column(String(20))
timezone = Column(String(50), default="America/New_York")
goal = Column(Enum(FitnessGoal), nullable=False)
status = Column(Enum(ClientStatus), default=ClientStatus.ACTIVE)
onboarding_date = Column(DateTime, default=datetime.utcnow)
# Coaching-specific fields
weekly_workout_target = Column(Integer, default=4)
preferred_communication = Column(String(20), default="email") # email, sms, both
checkin_frequency = Column(String(20), default="daily") # daily, weekly, biweekly
last_checkin_at = Column(DateTime)
# JSON blob for flexible custom data
custom_attributes = Column(JSONB, default=dict)
workouts = relationship("WorkoutLog", back_populates="client")
checkins = relationship("Checkin", back_populates="client")
flags = relationship("ClientFlag", back_populates="client")
class WorkoutLog(Base):
__tablename__ = "workout_logs"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
client_id = Column(UUID(as_uuid=True), ForeignKey("clients.id"), nullable=False)
completed_at = Column(DateTime, default=datetime.utcnow)
workout_type = Column(String(50)) # strength, cardio, mobility, etc.
duration_minutes = Column(Integer)
perceived_effort = Column(Integer) # RPE scale 1-10
notes = Column(Text)
metrics = Column(JSONB, default=dict) # e.g., {"sets": 5, "reps": 10, "weight_lbs": 185}
client = relationship("Client", back_populates="workouts")
class Checkin(Base):
__tablename__ = "checkins"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
client_id = Column(UUID(as_uuid=True), ForeignKey("clients.id"), nullable=False)
checkin_type = Column(String(30)) # automated, manual, missed
sent_at = Column(DateTime, default=datetime.utcnow)
response_received = Column(Boolean, default=False)
response_text = Column(Text)
ai_generated_message = Column(Text)
client = relationship("Client", back_populates="checkins")
class ClientFlag(Base):
__tablename__ = "client_flags"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
client_id = Column(UUID(as_uuid=True), ForeignKey("clients.id"), nullable=False)
flag_type = Column(String(50)) # adherence_drop, injury_report, plateau, disengagement
severity = Column(Integer) # 1-5 scale
created_at = Column(DateTime, default=datetime.utcnow)
resolved_at = Column(DateTime, nullable=True)
notes = Column(Text)
client = relationship("Client", back_populates="flags")Now the repository layer that the agent's tools will call:
# repositories/client_repo.py
from sqlalchemy import create_engine, func
from sqlalchemy.orm import sessionmaker
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from models.client import Client, WorkoutLog, Checkin, ClientFlag, ClientStatus
class ClientRepository:
"""Data access layer for client management operations."""
def __init__(self, database_url: str):
self.engine = create_engine(database_url)
self.Session = sessionmaker(bind=self.engine)
def get_client_by_id(self, client_id: str) -> Optional[Dict]:
session = self.Session()
try:
client = session.query(Client).filter(Client.id == client_id).first()
if not client:
return None
return self._client_to_dict(client)
finally:
session.close()
def get_clients_by_status(self, status: ClientStatus) -> List[Dict]:
session = self.Session()
try:
clients = session.query(Client).filter(Client.status == status).all()
return [self._client_to_dict(c) for c in clients]
finally:
session.close()
def get_recent_workouts(self, client_id: str, days: int = 14) -> List[Dict]:
session = self.Session()
try:
cutoff = datetime.utcnow() - timedelta(days=days)
workouts = session.query(WorkoutLog).filter(
WorkoutLog.client_id == client_id,
WorkoutLog.completed_at >= cutoff
).order_by(WorkoutLog.completed_at.desc()).all()
return [
{
"id": str(w.id),
"completed_at": w.completed_at.isoformat(),
"workout_type": w.workout_type,
"duration_minutes": w.duration_minutes,
"perceived_effort": w.perceived_effort,
"notes": w.notes,
"metrics": w.metrics
}
for w in workouts
]
finally:
session.close()
def get_adherence_rate(self, client_id: str, weeks: int = 4) -> float:
session = self.Session()
try:
client = session.query(Client).filter(Client.id == client_id).first()
if not client:
return 0.0
target_per_week = client.weekly_workout_target
total_target = target_per_week * weeks
cutoff = datetime.utcnow() - timedelta(weeks=weeks)
actual_count = session.query(func.count(WorkoutLog.id)).filter(
WorkoutLog.client_id == client_id,
WorkoutLog.completed_at >= cutoff
).scalar()
if total_target == 0:
return 100.0
return min((actual_count / total_target) * 100, 100.0)
finally:
session.close()
def create_checkin(self, client_id: str, message: str, checkin_type: str = "automated") -> Dict:
session = self.Session()
try:
checkin = Checkin(
client_id=client_id,
checkin_type=checkin_type,
ai_generated_message=message,
sent_at=datetime.utcnow()
)
session.add(checkin)
# Update last_checkin_at on client
client = session.query(Client).filter(Client.id == client_id).first()
if client:
client.last_checkin_at = datetime.utcnow()
session.commit()
return {"id": str(checkin.id), "sent_at": checkin.sent_at.isoformat()}
finally:
session.close()
def flag_client(self, client_id: str, flag_type: str, severity: int, notes: str) -> Dict:
session = self.Session()
try:
flag = ClientFlag(
client_id=client_id,
flag_type=flag_type,
severity=severity,
notes=notes
)
session.add(flag)
# Escalate client status if severity is high
if severity >= 4:
client = session.query(Client).filter(Client.id == client_id).first()
if client:
client.status = ClientStatus.AT_RISK
session.commit()
return {"id": str(flag.id), "flag_type": flag_type, "severity": severity}
finally:
session.close()
def _client_to_dict(self, client: Client) -> Dict:
return {
"id": str(client.id),
"first_name": client.first_name,
"last_name": client.last_name,
"email": client.email,
"phone": client.phone,
"timezone": client.timezone,
"goal": client.goal.value if client.goal else None,
"status": client.status.value if client.status else None,
"weekly_workout_target": client.weekly_workout_target,
"preferred_communication": client.preferred_communication,
"checkin_frequency": client.checkin_frequency,
"last_checkin_at": client.last_checkin_at.isoformat() if client.last_checkin_at else None,
"custom_attributes": client.custom_attributes
}Tool Definitions for the Agent
Tools are the agent's interface to the outside world. Each tool is a function with a clear docstring and typed parameters so the LLM knows exactly how to use it. Here are the essential tools for fitness coaching:
# tools/fitness_tools.py
from langchain.tools import tool
from typing import List, Optional, Dict
from datetime import datetime
from repositories.client_repo import ClientRepository
from services.messaging import MessageService
# These would be injected via a dependency container in production
client_repo = None # Set during application startup
message_service = None # Set during application startup
@tool
def get_client_profile(client_id: str) -> str:
"""Retrieve a client's full profile including goal, status, weekly targets, and history.
Use this tool whenever you need context about a specific client before making decisions.
Returns a structured summary of the client's information.
Args:
client_id: The UUID of the client to look up.
"""
client = client_repo.get_client_by_id(client_id)
if not client:
return f"Error: No client found with ID {client_id}"
recent_workouts = client_repo.get_recent_workouts(client_id, days=14)
adherence = client_repo.get_adherence_rate(client_id, weeks=4)
summary = f"""
Client Profile:
- Name: {client['first_name']} {client['last_name']}
- Goal: {client['goal']}
- Status: {client['status']}
- Weekly workout target: {client['weekly_workout_target']}
- 4-week adherence rate: {adherence:.1f}%
- Preferred contact: {client['preferred_communication']}
- Last check-in: {client.get('last_checkin_at', 'Never')}
- Recent workouts ({len(recent_workouts)} in past 14 days):
"""
for w in recent_workouts[:5]:
summary += f" ā¢ {w['completed_at'][:10]} - {w['workout_type']} ({w['duration_minutes']}min, RPE {w['perceived_effort']}/10)\n"
return summary
@tool
def send_client_message(client_id: str, message: str, channel: str = "auto") -> str:
"""Send a personalized message to a client via their preferred communication channel.
The message should be encouraging, specific, and reference their recent activity or goals.
Use the 'auto' channel to respect the client's preferred_communication setting.
Args:
client_id: The UUID of the client to message.
message: The exact message text to send. Keep under 320 characters for SMS compatibility.
channel: One of 'auto', 'email', or 'sms'. Default 'auto' uses client preference.
"""
client = client_repo.get_client_by_id(client_id)
if not client:
return f"Error: Client {client_id} not found"
if channel == "auto":
channel = client.get("preferred_communication", "email")
result = message_service.send(
to_email=client["email"],
to_phone=client.get("phone"),
message=message,
channel=channel
)
# Log the checkin
client_repo.create_checkin(client_id, message, checkin_type="automated")
return f"Message sent successfully to {client['first_name']} via {channel}. Delivery status: {result['status']}"
@tool
def analyze_adherence_trends(client_id: str) -> str:
"""Analyze a client's workout adherence patterns over recent weeks.
Use this to detect early signs of disengagement, plateaus, or overtraining.
Returns a detailed trend analysis with actionable insights.
Args:
client_id: The UUID of the client to analyze.
"""
# Get weekly adherence for the last 8 weeks
weekly_data = []
for week_offset in range(8):
week_start = datetime.utcnow().date() - datetime.timedelta(weeks=week_offset)
# Simplified: get workouts for that week
workouts = client_repo.get_recent_workouts(client_id, days=(week_offset + 1) * 7)
week_count = sum(1 for w in workouts if w["completed_at"] >= str(week_start))
weekly_data.append({"week_offset": week_offset, "count": week_count})
client = client_repo.get_client_by_id(client_id)
target = client["weekly_workout_target"]
analysis = f"8-week adherence trend (target: {target}/week):\n"
for wd in reversed(weekly_data):
pct = (wd["count"] / target * 100) if target > 0 else 0
bar = "ā" * min(int(pct / 10), 10)
analysis += f" Week -{wd['week_offset']}: {wd['count']}/{target} {bar} ({pct:.0f}%)\n"
# Detect concerning patterns
recent_avg = sum(w["count"] for w in weekly_data[:4]) / 4
earlier_avg = sum(w["count"] for w in weekly_data[4:]) / 4 if len(weekly_data) >= 8 else recent_avg
if recent_avg < earlier_avg * 0.7:
analysis += "\nā ļø SIGNIFICANT DROP: Recent 4-week average is substantially lower than prior period. Client may need re-engagement."
elif recent_avg > earlier_avg * 1.2:
analysis += "\nš IMPROVEMENT TREND: Recent adherence is notably higher. Positive momentum."
return analysis
@tool
def flag_client_for_coach_review(client_id: str, concern_type: str, severity: int, detailed_notes: str) -> str:
"""Create a flagged alert that requires the human coach's attention.
Use this for serious concerns like sustained disengagement, reported injuries,
or major goal deviations. Severity 1-3 is informational, 4-5 requires immediate action.
Args:
client_id: The UUID of the client.
concern_type: Category like 'adherence_drop', 'injury_report', 'plateau', 'disengagement'.
severity: Integer 1-5, where 4+ triggers client status change to 'at_risk'.
detailed_notes: Full context explaining why this flag was raised.
"""
result = client_repo.flag_client(client_id, concern_type, severity, detailed_notes)
return f"Flag created: {result['flag_type']} (severity {severity}). Coach review required. Client status: {'escalated to at_risk' if severity >= 4 else 'unchanged'}"
@tool
def query_workout_history(client_id: str, days: int = 30) -> str:
"""Retrieve detailed workout history for a client over a specified period.
Use this when you need specific workout data to provide feedback or analyze patterns.
Args:
client_id: The UUID of the client.
days: Number of days to look back (default 30).
"""
workouts = client_repo.get_recent_workouts(client_id, days=days)
if not workouts:
return f"No workouts found for client {client_id} in the past {days} days."
result = f"Workout history (past {days} days, {len(workouts)} sessions):\n"
for w in workouts:
metrics_str = ", ".join(f"{k}: {v}" for k, v in w.get("metrics", {}).items())
result += f" [{w['completed_at'][:10]}] {w['workout_type']} - {w['duration_minutes']}min, RPE {w['perceived_effort']}/10"
if metrics_str:
result += f" | {metrics_str}"
if w.get("notes"):
result += f" | Notes: {w['notes'][:80]}"
result += "\n"
return result
@tool
def get_clients_needing_attention() -> str:
"""Get a list of all clients that may need attention based on status and recent activity.
Use this for daily monitoring runs to identify who needs a check-in or coach intervention.
Returns a prioritized list.
"""
at_risk = client_repo.get_clients_by_status(ClientStatus.AT_RISK)
active = client_repo.get_clients_by_status(ClientStatus.ACTIVE)
result = "Clients Needing Attention:\n\n"
if at_risk:
result += f"š“ AT RISK ({len(at_risk)} clients):\n"
for c in at_risk:
adherence = client_repo.get_adherence_rate(str(c["id"]), weeks=2)
result += f" ā¢ {c['first_name']} {c['last_name']} - {c['goal']} - 2wk adherence: {adherence:.0f}% - Last checkin: {c.get('last_checkin_at', 'N/A')}\n"
# Active clients with low recent adherence
result += "\nš” ACTIVE - LOW ENGAGEMENT:\n"
for c in active:
adherence = client_repo.get_adherence_rate(str(c["id"]), weeks=2)
if adherence < 50:
result += f" ā¢ {c['first_name']} {c['last_name']} - {c['goal']} - 2wk adherence: {adherence:.0f}%\n"
return result if "ā¢" in result else result + "All clients are currently on track. Great job!"
@tool
def schedule_follow_up(client_id: str, scheduled_for: str, reason: str) -> str:
"""Schedule a follow-up task for the human coach with a specific client.
Use this when the agent determines a client needs direct coach intervention
that cannot be handled automatically (e.g., program redesign, injury consultation).
Args:
client_id: The UUID of the client.
scheduled_for: ISO-formatted datetime string for when the follow-up should occur.
reason: Clear explanation of why the follow-up is needed.
"""
# In production, this would integrate with a calendar API or task queue
# For now, we log it and return confirmation
client = client_repo.get_client_by_id(client_id)
return f"Follow-up scheduled for {client['first_name']} {client['last_name']} on {scheduled_for}. Reason: {reason}. Coach will be notified."Automated Check-in System
The daily check-in workflow is the most common agent task. Here is the orchestration service that runs periodically (via Celery or a cron-triggered endpoint) and processes all clients:
# services/checkin_service.py
from agent.core import FitnessCoachAgent
from tools.fitness_tools import (
get_client_profile, send_client_message, analyze_adherence_trends,
flag_client_for_coach_review, query_workout_history,
get_clients_needing_attention, schedule_follow_up
)
from repositories.client_repo import ClientRepository
from datetime import datetime
from typing import List, Dict
import logging
logger = logging.getLogger(__name__)
class CheckinOrchestrator:
"""Orchestrates the daily automated check-in workflow across all clients."""
SYSTEM_PROMPT = """You are a compassionate and observant fitness coaching AI assistant.
Your job is to review client data and decide on the appropriate action for each client.
Rules:
1. If a client has worked out in the past 24 hours, send a brief encouraging message
that references their specific workout (type, duration, or achievement).
2. If a client has NOT worked out in 48+ hours and their adherence is below 60%,
send a gentle check-in message. Do NOT guilt them ā be supportive.
3. If adherence has dropped more than 30% over the past 2 weeks compared to prior period,
flag the client for coach review at severity 3.
4. If a client reports pain or injury in their workout notes, flag immediately at severity 5.
5. If the client is marked 'at_risk' and has had no activity for 7+ days,
schedule a coach follow-up within 48 hours.
6. Always personalize messages using the client's name and reference their specific goal.
7. Keep SMS messages under 320 characters. Email messages can be longer but should be warm and concise.
8. If everything looks good (recent workout, high adherence), you may skip the client
or send a quick positive message.
Before acting, always use get_client_profile to understand the client's full context.
After sending a message, log it via the tool ā it records the check-in automatically."""
def __init__(self, database_url: str):
self.repo = ClientRepository(database_url)
tools = [
get_client_profile,
send_client_message,
analyze_adherence_trends,
flag_client_for_coach_review,
query_workout_history,
get_clients_needing_attention,
schedule_follow_up
]
self.agent = FitnessCoachAgent(tools=tools, system_prompt=self.SYSTEM_PROMPT)
async def process_single_client(self, client_id: str) -> Dict:
"""Run the agent on a single client and return the action taken."""
context = {
"current_time": datetime.utcnow().isoformat(),
"chat_history": []
}
input_text = f"""
Please review client {client_id} and take appropriate action.
Start by getting their full profile, then analyze their recent adherence.
Decide whether to send a message, flag for review, or schedule a follow-up.
Take exactly ONE primary action per run (send message OR flag OR schedule follow-up).
Client ID: {client_id}
"""
try:
result = await self.agent.run(input_text, context)
return {
"client_id": client_id,
"action_taken": result["output"],
"tool_calls": [
str(step) for step in result.get("intermediate_steps", [])
],
"success": True
}
except Exception as e:
logger.error(f"Agent failed for client {client_id}: {e}")
return {"client_id": client_id, "success": False, "error": str(e)}
async def run_daily_checkin_batch(self, client_ids: List[str] = None) -> List[Dict]:
"""Process all specified clients (or all active/at-risk clients if none specified)."""
if client_ids is None:
# Get all clients that need processing
active = self.repo.get_clients_by_status("active")
at_risk = self.repo.get_clients_by_status("at_risk")
client_ids = [c["id"] for c in active + at_risk]
results = []
for cid in client_ids:
result = await self.process_single_client(cid)
results.append(result)
logger.info(f"Processed {cid}: {result.get('action_taken', 'failed')[:100]}")