Integrating Human Input Back to Agents

Integrating Human Input Back to Agents | CodeSignal Learn

Introduction: Completing the Human-in-the-Loop Workflow

In the previous lesson, you completed Factor 6 by implementing pause and resume endpoints that give clients lifecycle control over agent workflows . Now, you'll implement Factor 7 — Contact humans with tool calls . As you learned when studying the 12-Factor Agents methodology, this factor treats human escalation as a first-class tool: when the agent lacks information, it doesn't guess or fail — it calls a structured ask_human tool, pauses itself, and waits for a response. The human's input is recorded in the context just like any other tool output, making the interaction auditable and reproducible . By the end of this lesson, you'll have a complete human-in-the-loop system where agents and users collaborate seamlessly to solve problems together.

Defining the ask_human Tool Schema

To allow the agent to request information from users, you need to create a new tool schema. The ask_human tool requires only one parameter: the question or prompt that the agent wants to present to the user. Create a new file at src/core/tools/schemas/ask_human.json with the following content: JSON{ "type": "function", "name": "ask_human", "description": "Ask the user for clarification or additional information.", "strict": true, "parameters": { "type": "object", "properties": { "question": { "type": "string", "description": "The question or prompt to ask the user" } }, "required": ["question"], "additionalProperties": false } }{ "type": "function", "name": "ask_human", "description": "Ask the user for clarification or additional information.", "strict": true, "parameters": { "type": "object", "properties": { "question": { "type": "string", "description": "The question or prompt to ask the user" } }, "required": ["question"], "additionalProperties": false } } This schema follows the same structure as your other tools, like sum_numbers or final_answer. The name field identifies the function as ask_human, while the description tells the language model when it should use this tool. The parameters section defines a single required field called question that holds the text the agent wants to show to the user. By keeping the schema simple with just one string parameter, you make it easy for the language model to formulate clear questions without worrying about complex argument structures.

Loading the ask_human Schema in the Agent

Join the 1M+ learners on CodeSignal

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

Handling ask_human in the Agent's Step Logic

Creating the provide_input Endpoint Request Model

To accept human responses and resume agent execution, you need to create a new API endpoint . Start by defining a Pydantic model for the request payload in src/server/main.py : Python class ProvideInputRequest(BaseModel): id: str answer: str class ProvideInputRequest(BaseModel): id: str answer: str This model requires two fields: id identifies which agent state should receive the input, and answer contains the human's response to the agent's question.

Implementing the provide_input Route Handler

Now implement the route handler that validates the request and prepares the state for resumption: Python@app.post("/agent/provide_input", response_model=State) def provide_input(payload: ProvideInputRequest, background_tasks: BackgroundTasks): """Provide human input to a state waiting for human input and resume execution""" 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") # Check status while still in session if db_state.status != "waiting_human_input": raise HTTPException( status_code=400, detail=f"State is not waiting for human input. Current status: {db_state.status}" ) # Convert to Pydantic while still in session working_state = db_to_pydantic(db_state)@app.post("/agent/provide_input", response_model=State) def provide_input(payload: ProvideInputRequest, background_tasks: BackgroundTasks): """Provide human input to a state waiting for human input and resume execution""" 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") # Check status while still in session if db_state.status != "waiting_human_input": raise HTTPException( status_code=400, detail=f"State is not waiting for human input. Current status: {db_state.status}" ) # Convert to Pydantic while still in session working_state = db_to_pydantic(db_state) The endpoint opens a database session and queries for the state record matching the provided id, returning a 404 error if it does not exist. Before accepting the input, the handler validates that the state's current status is waiting_human_input. This prevents clients from accidentally sending answers to agents that are not waiting for them. After validation passes, the code converts the database record to a Pydantic model to work with in the subsequent logic.

Extracting the call_id from the Context

Before you can add the human's answer to the agent's context, you need to determine which specific ask_human call the answer corresponds to. The language model requires this information because it tracks function calls and their outputs using unique call_id values. For that, let's implement a module-level helper function in src/server/main.py that searches backwards through the context: Python def _get_call_id_from_state(state: State) -> Optional[str]: """Extract the call_id from the last ask_human call in context""" # Search backwards through context to find the most recent ask_human call # We need the call_id to match the human's response with the original call for item in reversed(state.context): if isinstance(item, dict) and item.get("type") == "function_call" and item.get("name") == "ask_human": return item.get("call_id") return None def _get_call_id_from_state(state: State) -> Optional[str]: """Extract the call_id from the last ask_human call in context""" # Search backwards through context to find the most recent ask_human call # We need the call_id to match the human's response with the original call for item in reversed(state.context): if isinstance(item, dict) and item.get("type") == "function_call" and item.get("name") == "ask_human": return item.get("call_id") return None We will place this function above the provide_input route handler so it's available when the endpoint calls it. The function iterates through the context in reverse order using Python's reversed() function, which is more efficient than searching from the beginning when the ask_human call is likely near the end. For each item, it checks whether the item is a dictionary with type equal to function_call and name equal to ask_human . When it finds such an item, it extracts and returns the call_id field. If no matching call is found, the function returns None .

Using the call_id to Validate the Context

Now, back inside the provide_input route handler, you can call this helper to retrieve the call_id needed for constructing the response: Python # Find the call_id from the last ask_human call # This matches the human's answer to the original ask_human function call call_id = _get_call_id_from_state(working_state) if not call_id: raise HTTPException(status_code=400, detail="Could not find ask_human call in state context") # Find the call_id from the last ask_human call # This matches the human's answer to the original ask_human function call call_id = _get_call_id_from_state(working_state) if not call_id: raise HTTPException(status_code=400, detail="Could not find ask_human call in state context") After extracting the call_id, the code validates that it exists. If_get_call_id_from_state returns None, the endpoint raises a 400 error, indicating that the context is in an invalid state.

Constructing the Function Call Output and Securing the Lifecycle

With the call_id identified, you can construct the function_call_output structure that the language model expects and append it to the agent's context. Python# Add the human's answer as a function_call_output to context # Format matches what the Responses API expects human_response = { "type": "function_call_output", "call_id": call_id, # Match with the original ask_human call "output": json.dumps({"answer": payload.answer}) # Wrap answer in JSON } working_state.context.append(human_response) # Update state in database with new context with get_db_session() as session: db_state = session.query(StateModel).filter(StateModel.id == payload.id).first() db_state.context = working_state.context db_state.status = "running" # Change status back to running so agent can continue # Run agent in background (non-blocking) background_tasks.add_task(_run_agent_in_background, payload.id, working_state) return working_state# Add the human's answer as a function_call_output to context # Format matches what the Responses API expects human_response = { "type": "function_call_output", "call_id": call_id, # Match with the original ask_human call "output": json.dumps({"answer": payload.answer}) # Wrap answer in JSON } working_state.context.append(human_response) # Update state in database with new context with get_db_session() as session: db_state = session.query(StateModel).filter(StateModel.id == payload.id).first() db_state.context = working_state.context db_state.status = "running" # Change status back to running so agent can continue # Run agent in background (non-blocking) background_tasks.add_task(_run_agent_in_background, payload.id, working_state) return working_state The human_response dictionary links the output to the original ask_human call using the extracted call_id, and wraps the human's response in a JSON string. The endpoint then sets the status to running and schedules the background task, allowing the agent to continue seamlessly. Important Note: Because the agent is now waiting for specific input, you must also secure your existing /agent/resume endpoint. If a client blindly calls resume while the agent is waiting for a human, the agent will loop indefinitely or fail. You should update /agent/resume to reject requests with a 400 error if the state is waiting_human_input, forcing clients to use /agent/provide_input instead. You'll do this in the upcoming exercise!

Testing the Launch and Waiting Phase

To verify that the human-in-the-loop workflow functions correctly, let's write a test script that launches an agent with an incomplete prompt and waits for the ask_human call. Pythonimport time import json import requests BASE_URL = "http://localhost:8000" # Launch a new agent that will likely ask for one missing value (radius/diameter) response = requests.post( f"{BASE_URL}/agent/launch", json={"input_prompt": "Can you figure out the area of a circle for me? Use pi = 3.14."} ) state = response.json() print(f"Launched agent with ID: {state['id']}") # Poll until it's waiting for human input 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"] == "waiting_human_input": question = None for item in reversed(current_state.get("context", [])): if ( isinstance(item, dict) and item.get("type") == "function_call" and item.get("name") == "ask_human" ): args = item.get("arguments") try: args_dict = json.loads(args) if isinstance(args, str) else (args or {}) except json.JSONDecodeError: args_dict = {} question = args_dict.get("question") break print(f"Question: {question}") break if current_state["status"] in ["complete", "max_steps_reached", "failed"]: print(f"Completed: {current_state.get('final_answer')}") break time.sleep(1)import time import json import requests BASE_URL = "http://localhost:8000" # Launch a new agent that will likely ask for one missing value (radius/diameter) response = requests.post( f"{BASE_URL}/agent/launch", json={"input_prompt": "Can you figure out the area of a circle for me? Use pi = 3.14."} ) state = response.json() print(f"Launched agent with ID: {state['id']}") # Poll until it's waiting for human input 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"] == "waiting_human_input": question = None for item in reversed(current_state.get("context", [])): if ( isinstance(item, dict) and item.get("type") == "function_call" and item.get("name") == "ask_human" ): args = item.get("arguments") try: args_dict = json.loads(args) if isinstance(args, str) else (args or {}) except json.JSONDecodeError: args_dict = {} question = args_dict.get("question") break print(f"Question: {question}") break if current_state["status"] in ["complete", "max_steps_reached", "failed"]: print(f"Completed: {current_state.get('final_answer')}") break time.sleep(1) Running this script produces output showing the agent launching and running until it needs human input: textLaunched agent with ID: cb0636b7-003c-47ff-9c0d-2f0fb791a4bf Status: running, Steps: 0 Status: running, Steps: 1 Status: running, Steps: 2 Status: waiting_human_input, Steps: 2 Question: What is the circle's size? Please provide either the radius or the diameter (and specify which one), and I'll compute the area using pi = 3.14.Launched agent with ID: cb0636b7-003c-47ff-9c0d-2f0fb791a4bf Status: running, Steps: 0 Status: running, Steps: 1 Status: running, Steps: 2 Status: waiting_human_input, Steps: 2 Question: What is the circle's size? Please provide either the radius or the diameter (and specify which one), and I'll compute the area using pi = 3.14. The output shows that the agent launched successfully and ran for two steps before calling ask_human to request the missing circle measurement. The question it formulated is clear and specific, asking for either radius or diameter and mentioning that it will use pi equals three point one four as specified in the original prompt.

Testing the Input and Completion Phase

Now extend the test script to provide the human input and monitor completion: Python# Provide human input if current_state["status"] == "waiting_human_input": requests.post( f"{BASE_URL}/agent/provide_input", json={"id": state["id"], "answer": "The radius is 10 inches."} ) print("Provided input, agent resuming...") # 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"Final answer: {current_state.get('final_answer')}") break time.sleep(1)# Provide human input if current_state["status"] == "waiting_human_input": requests.post( f"{BASE_URL}/agent/provide_input", json={"id": state["id"], "answer": "The radius is 10 inches."} ) print("Provided input, agent resuming...") # 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"Final answer: {current_state.get('final_answer')}") break time.sleep(1) Running this continuation of the script produces output demonstrating the complete human-in-the-loop cycle: textProvided input, agent resuming... Status: running, Steps: 2 Status: running, Steps: 3 Status: running, Steps: 4 Status: running, Steps: 5 Status: running, Steps: 6 Status: complete, Steps: 6 Final answer: The area of the circle is 314.0 square inches (using π = 3.14 and radius = 10 inches).Provided input, agent resuming... Status: running, Steps: 2 Status: running, Steps: 3 Status: running, Steps: 4 Status: running, Steps: 5 Status: running, Steps: 6 Status: complete, Steps: 6 Final answer: The area of the circle is 314.0 square inches (using π = 3.14 and radius = 10 inches). After the script provides the radius of ten inches, the agent resumes execution and continues for four more steps, ultimately reaching completion with the correct calculated area of 314.0 square inches. If you examine the full context in the final state, you'll see the ask_human function call followed by the function_call_output containing the human's answer — both linked by the same call_id, providing a complete audit trail of the collaboration.

Integrating Human Input Back to Agents

Introduction: Completing the Human-in-the-Loop Workflow

Defining the ask_human Tool Schema

Loading the ask_human Schema in the Agent

Join the 1M+ learners on CodeSignal

Handling ask_human in the Agent's Step Logic

Creating the provide_input Endpoint Request Model

Implementing the provide_input Route Handler

Extracting the call_id from the Context

Using the call_id to Validate the Context

Constructing the Function Call Output and Securing the Lifecycle

Testing the Launch and Waiting Phase

Testing the Input and Completion Phase

Summary