Pausing and Resuming Agents Through API Calls

Introduction: Adding Lifecycle Control to Agent APIs

In the previous lesson, you implemented database persistence and progress callbacks that save the agent's state after each step. This allows clients to monitor progress in real time, but they still have no control over the agent once it starts running. Now it's time to complete Factor 6 — Launch / Pause / Resume with simple APIs . You already have the launch endpoint from lesson one and the durable persistence layer from lesson two; in this lesson, you add the pause and resume endpoints that give users the power to stop expensive computations temporarily or correct mistakes without losing progress. You will implement these endpoints using a cooperative pause mechanism that lets agents stop gracefully between steps.

How the Pause Mechanism Works

Unlike forcefully terminating a thread or process, a cooperative pause lets the agent finish its current step before stopping. This approach works by having the pause endpoint change the database status field to "paused" and requiring the agent's progress callback to check for this status change after each step completes. When the callback detects that the status has been changed externally, it updates the local state object to reflect the pause, causing the agent's main loop to exit on its next iteration. The advantage of this design is that it never interrupts the agent mid-step, ensuring the agent's state remains consistent and valid, with all fields properly synchronized. This is exactly what Factor 6 calls for: the agent can checkpoint, wait, and resume reliably across time.

Modifying the Progress Callback to Detect Pause Requests

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

def _create_progress_callback(state_id: str):
    """Create a progress callback function that saves state after each step"""
    def save_progress(state: State):
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if db_state:
                # Check if status was changed to "paused" externally (via pause endpoint)
                if db_state.status == "paused":
                    # Update local state to paused so agent loop will exit
                    state.status = "paused"
                    # Don't overwrite the paused status - just save other fields
                    db_state.steps = state.steps
                    db_state.context = state.context
                    db_state.pending_tool_calls = state.pending_tool_calls
                    db_state.error = state.error
                    db_state.final_answer = state.final_answer
                else:
                    # Normal save - update all fields including status
                    db_state.steps = state.steps
                    db_state.status = state.status
                    db_state.context = state.context
                    db_state.pending_tool_calls = state.pending_tool_calls
                    db_state.error = state.error
                    db_state.final_answer = state.final_answer
    return save_progress

Python

class PauseRequest(BaseModel):
    id: str

class PauseRequest(BaseModel):
    id: str

Python

@app.post("/agent/pause", response_model=State)
def agent_pause(payload: PauseRequest):
    """Pause a running agent workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Only allow pausing if agent is currently running
        if db_state.status != "running":
            raise HTTPException(
                status_code=400,
                detail=f"Cannot pause agent. Current status: {db_state.status}"
            )
        
        # Set status to paused (progress callback will detect this and stop the loop)
        db_state.status = "paused"
        
        return db_to_pydantic(db_state)

@app.post("/agent/pause", response_model=State)
def agent_pause(payload: PauseRequest):
    """Pause a running agent workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Only allow pausing if agent is currently running
        if db_state.status != "running":
            raise HTTPException(
                status_code=400,
                detail=f"Cannot pause agent. Current status: {db_state.status}"
            )
        
        # Set status to paused (progress callback will detect this and stop the loop)
        db_state.status = "paused"
        
        return db_to_pydantic(db_state)

Python

def _run_agent_in_background(state_id: str, working_state: Optional[State] = None):
    """Run the agent in a background thread and update the database"""
    # If working_state not provided, load from database
    if working_state is None:
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if not db_state:
                return
            db_state.status = "running"
            working_state = db_to_pydantic(db_state)
    else:
        # working_state provided (for resume), ensure DB status is running
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if db_state:
                db_state.status = "running"
    
    # Run agent with progress callback
    save_progress = _create_progress_callback(state_id)
    final_state = agent.run(working_state, progress_callback=save_progress)
    
    # Final update
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
        if db_state:
            db_state.steps = final_state.steps
            db_state.status = final_state.status
            db_state.context = final_state.context
            db_state.pending_tool_calls = final_state.pending_tool_calls
            db_state.error = final_state.error
            db_state.final_answer = final_state.final_answer

def _run_agent_in_background(state_id: str, working_state: Optional[State] = None):
    """Run the agent in a background thread and update the database"""
    # If working_state not provided, load from database
    if working_state is None:
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if not db_state:
                return
            db_state.status = "running"
            working_state = db_to_pydantic(db_state)
    else:
        # working_state provided (for resume), ensure DB status is running
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if db_state:
                db_state.status = "running"
    
    # Run agent with progress callback
    save_progress = _create_progress_callback(state_id)
    final_state = agent.run(working_state, progress_callback=save_progress)
    
    # Final update
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
        if db_state:
            db_state.steps = final_state.steps
            db_state.status = final_state.status
            db_state.context = final_state.context
            db_state.pending_tool_calls = final_state.pending_tool_calls
            db_state.error = final_state.error
            db_state.final_answer = final_state.final_answer

Python

class ResumeRequest(BaseModel):
    id: str

class ResumeRequest(BaseModel):
    id: str

Python

@app.post("/agent/resume", response_model=State)
def agent_resume(payload: ResumeRequest, background_tasks: BackgroundTasks):
    """Resume a paused or interrupted workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Prevent concurrent execution
        if db_state.status == "running":
            raise HTTPException(status_code=409, detail="Agent is already running")
            
        # Set status to running so the response reflects reality
        db_state.status = "running"

        # Convert to Pydantic
        working_state = db_to_pydantic(db_state)
    
    # Run agent in background (non-blocking)
    background_tasks.add_task(_run_agent_in_background, payload.id, working_state)
    
    return working_state

@app.post("/agent/resume", response_model=State)
def agent_resume(payload: ResumeRequest, background_tasks: BackgroundTasks):
    """Resume a paused or interrupted workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Prevent concurrent execution
        if db_state.status == "running":
            raise HTTPException(status_code=409, detail="Agent is already running")
            
        # Set status to running so the response reflects reality
        db_state.status = "running"

        # Convert to Pydantic
        working_state = db_to_pydantic(db_state)
    
    # Run agent in background (non-blocking)
    background_tasks.add_task(_run_agent_in_background, payload.id, working_state)
    
    return working_state

Python

import time
import requests

BASE_URL = "http://localhost:8000"

# Launch a new agent
response = requests.post(
    f"{BASE_URL}/agent/launch",
    json={"input_prompt": "Solve the root of this equation: x^2 - 5x + 6 = 0"}
)
state = response.json()
print(f"Launched agent with ID: {state['id']}")

# Wait a bit
time.sleep(3)

# Pause the agent
response = requests.post(
    f"{BASE_URL}/agent/pause",
    json={"id": state['id']}
)
paused_state = response.json()
print(f"Status: {paused_state['status']} | Steps: {paused_state['steps']} | Pending tool calls: {paused_state['pending_tool_calls']}")

import time
import requests

BASE_URL = "http://localhost:8000"

# Launch a new agent
response = requests.post(
    f"{BASE_URL}/agent/launch",
    json={"input_prompt": "Solve the root of this equation: x^2 - 5x + 6 = 0"}
)
state = response.json()
print(f"Launched agent with ID: {state['id']}")

# Wait a bit
time.sleep(3)

# Pause the agent
response = requests.post(
    f"{BASE_URL}/agent/pause",
    json={"id": state['id']}
)
paused_state = response.json()
print(f"Status: {paused_state['status']} | Steps: {paused_state['steps']} | Pending tool calls: {paused_state['pending_tool_calls']}")

text

Launched agent with ID: e377469b-4b33-4327-9e9d-c1288d133201
Status: paused | Steps: 0 | Pending tool calls: []

Launched agent with ID: e377469b-4b33-4327-9e9d-c1288d133201
Status: paused | Steps: 0 | Pending tool calls: []

Python

# Resume the agent
response = requests.post(
    f"{BASE_URL}/agent/resume",
    json={"id": state['id']}
)
resumed_state = response.json()
print(f"Resumed agent. Status: {resumed_state['status']}")

# Poll for completion
while True:
    response = requests.get(f"{BASE_URL}/agent/state/{state['id']}")
    current_state = response.json()
    print(f"Status: {current_state['status']} | Steps: {current_state['steps']}")
    
    if current_state['status'] in ["complete", "max_steps_reached", "failed"]:
        print(f"\nFinal answer: {current_state.get('final_answer')}")
        break
    
    time.sleep(1)

# Resume the agent
response = requests.post(
    f"{BASE_URL}/agent/resume",
    json={"id": state['id']}
)
resumed_state = response.json()
print(f"Resumed agent. Status: {resumed_state['status']}")

# Poll for completion
while True:
    response = requests.get(f"{BASE_URL}/agent/state/{state['id']}")
    current_state = response.json()
    print(f"Status: {current_state['status']} | Steps: {current_state['steps']}")
    
    if current_state['status'] in ["complete", "max_steps_reached", "failed"]:
        print(f"\nFinal answer: {current_state.get('final_answer')}")
        break
    
    time.sleep(1)

text

Resumed agent. Status: running
Status: running | Steps: 0
...
Status: complete | Steps: 6

Final answer: The roots are x = 2 and x = 3.

Resumed agent. Status: running
Status: running | Steps: 0
...
Status: complete | Steps: 6

Final answer: The roots are x = 2 and x = 3.

Python

class PauseRequest(BaseModel):
    id: str

Python

@app.post("/agent/pause", response_model=State)
def agent_pause(payload: PauseRequest):
    """Pause a running agent workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Only allow pausing if agent is currently running
        if db_state.status != "running":
            raise HTTPException(
                status_code=400,
                detail=f"Cannot pause agent. Current status: {db_state.status}"
            )
        
        # Set status to paused (progress callback will detect this and stop the loop)
        db_state.status = "paused"
        
        return db_to_pydantic(db_state)

Python

def _run_agent_in_background(state_id: str, working_state: Optional[State] = None):
    """Run the agent in a background thread and update the database"""
    # If working_state not provided, load from database
    if working_state is None:
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if not db_state:
                return
            db_state.status = "running"
            working_state = db_to_pydantic(db_state)
    else:
        # working_state provided (for resume), ensure DB status is running
        with get_db_session() as session:
            db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
            if db_state:
                db_state.status = "running"
    
    # Run agent with progress callback
    save_progress = _create_progress_callback(state_id)
    final_state = agent.run(working_state, progress_callback=save_progress)
    
    # Final update
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == state_id).first()
        if db_state:
            db_state.steps = final_state.steps
            db_state.status = final_state.status
            db_state.context = final_state.context
            db_state.pending_tool_calls = final_state.pending_tool_calls
            db_state.error = final_state.error
            db_state.final_answer = final_state.final_answer

Python

class ResumeRequest(BaseModel):
    id: str

Python

@app.post("/agent/resume", response_model=State)
def agent_resume(payload: ResumeRequest, background_tasks: BackgroundTasks):
    """Resume a paused or interrupted workflow"""
    with get_db_session() as session:
        db_state = session.query(StateModel).filter(StateModel.id == payload.id).first()
        if not db_state:
            raise HTTPException(status_code=404, detail="State not found")
        
        # Prevent concurrent execution
        if db_state.status == "running":
            raise HTTPException(status_code=409, detail="Agent is already running")
            
        # Set status to running so the response reflects reality
        db_state.status = "running"

        # Convert to Pydantic
        working_state = db_to_pydantic(db_state)
    
    # Run agent in background (non-blocking)
    background_tasks.add_task(_run_agent_in_background, payload.id, working_state)
    
    return working_state

Python

import time
import requests

BASE_URL = "http://localhost:8000"

# Launch a new agent
response = requests.post(
    f"{BASE_URL}/agent/launch",
    json={"input_prompt": "Solve the root of this equation: x^2 - 5x + 6 = 0"}
)
state = response.json()
print(f"Launched agent with ID: {state['id']}")

# Wait a bit
time.sleep(3)

# Pause the agent
response = requests.post(
    f"{BASE_URL}/agent/pause",
    json={"id": state['id']}
)
paused_state = response.json()
print(f"Status: {paused_state['status']} | Steps: {paused_state['steps']} | Pending tool calls: {paused_state['pending_tool_calls']}")

text

Launched agent with ID: e377469b-4b33-4327-9e9d-c1288d133201
Status: paused | Steps: 0 | Pending tool calls: []

Python

# Resume the agent
response = requests.post(
    f"{BASE_URL}/agent/resume",
    json={"id": state['id']}
)
resumed_state = response.json()
print(f"Resumed agent. Status: {resumed_state['status']}")

# Poll for completion
while True:
    response = requests.get(f"{BASE_URL}/agent/state/{state['id']}")
    current_state = response.json()
    print(f"Status: {current_state['status']} | Steps: {current_state['steps']}")
    
    if current_state['status'] in ["complete", "max_steps_reached", "failed"]:
        print(f"\nFinal answer: {current_state.get('final_answer')}")
        break
    
    time.sleep(1)

text

Resumed agent. Status: running
Status: running | Steps: 0
...
Status: complete | Steps: 6

Final answer: The roots are x = 2 and x = 3.

Introduction: Adding Lifecycle Control to Agent APIs

How the Pause Mechanism Works

Modifying the Progress Callback to Detect Pause Requests

Join the 1M+ learners on CodeSignal

Implementing the Pause Endpoint

Creating the Pause Route Handler

Adapting the Background Runner for Resume Support

Implementing the Resume Endpoint

Creating the Resume Route Handler

Testing Launch and Pause

Testing Resume and Monitoring Completion

Summary and Next Steps