Deploying Agent Workflows with Human-in-the-Loop and Function Calling in FastAPI over Websockets

{'globals': {'agents': '["Agent"]',
             'current_agent_name': '"Agent"',
             'handoff_output_prompt': '"Agent {to_agent} is now handling the '
                                      'request due to the following reason: '
                                      '{reason}.\\nPlease continue with the '
                                      'current request."',
             'memory': '{"__is_component": true, "value": {"chat_store": '
                       '{"store": {"chat_history": [{"role": "user", '
                       '"additional_kwargs": {}, "blocks": [{"block_type": '
                       '"text", "text": "I want to proceed with the dangerous '
                       'task."}]}]}, "class_name": "SimpleChatStore"}, '
                       '"chat_store_key": "chat_history", "token_limit": '
                       '96000, "class_name": "ChatMemoryBuffer"}, '
                       '"qualified_name": '
                       '"llama_index.core.memory.chat_memory_buffer.ChatMemoryBuffer"}',
             'num_tool_calls': '1',
             'scratchpad': '[{"__is_pydantic": true, "value": {"role": '
                           '"assistant", "additional_kwargs": {"tool_calls": '
                           '[{"index": 0, "id": '
                           '"call_gASClWjTwjd45RkalhzwruvQ", "function": '
                           '{"arguments": "{}", "name": "dangerous_task"}, '
                           '"type": "function"}]}, "blocks": [{"block_type": '
                           '"text", "text": ""}]}, "qualified_name": '
                           '"llama_index.core.base.llms.types.ChatMessage"}]',
             'state': '{}',
             'user_msg_str': '"I want to proceed with the dangerous task."'},
 'in_progress': {'call_tool': ['{"__is_pydantic": true, "value": {"tool_name": '
                               '"dangerous_task", "tool_kwargs": {}, '
                               '"tool_id": "call_gASClWjTwjd45RkalhzwruvQ"}, '
                               '"qualified_name": '
                               '"llama_index.core.agent.workflow.workflow_events.ToolCall"}'],
                 'init_run': [],
                 'parse_agent_output': [],
                 'run_agent_step': [],
                 'setup_agent': []},
 'is_running': True,
 'queues': {'32fe8d6c-4699-42a3-8dad-9d76cfdc19c8': '["{\\"__is_pydantic\\": ' ….”]’,
            '_done': '[]',
            'aggregate_tool_results': '[]',
            'call_tool': '[]',
            'init_run': '[]',
            'parse_agent_output': '[]',
            'run_agent_step': '[]',
            'setup_agent': '[]'},
 'stepwise': False,
 'streaming_queue': '[]',
 'waiter_id': '32fe8d6c-4699-42a3-8dad-9d76cfdc19c8'}

from llama_index.llms.openai import OpenAI
import asyncio
llm = OpenAI(model="gpt-4o-mini", api_key="sk-...")

from llama_index.core.workflow import (
    Context,
    InputRequiredEvent,
    HumanResponseEvent,
)
from llama_index.core.agent.workflow import AgentWorkflow


async def dangerous_task(ctx: Context) -> str:
    """A dangerous task that requires human confirmation."""
    ctx.write_event_to_stream(
        InputRequiredEvent(
            prefix="Are you sure you want to proceed?",
            user_name="Logan",
        )
    )

    response = await ctx.wait_for_event(
        HumanResponseEvent, requirements={"user_name": "Logan"}
    )
    if response.response == "yes":
        return "Dangerous task completed successfully."
    else:
        return "Dangerous task aborted."


async def main():
    workflow = AgentWorkflow.from_tools_or_functions(
        [dangerous_task],
        llm=llm,
        system_prompt="You are a helpful assistant that can perform dangerous tasks.",
        verbose=True,
    )

    from llama_index.core.workflow import JsonSerializer

    handler = workflow.run(user_msg="I want to proceed with the dangerous task.")

    input_ev = None
    ctx_dict = None
    async for event in handler.stream_events():
        if isinstance(event, InputRequiredEvent):
            input_ev = event  

            # save the context somewhere for later
            ctx_dict = handler.ctx.to_dict(serializer=JsonSerializer())
            break
    
    import pprint
    pprint.pprint(ctx_dict)



    # get the response from the user
    response_str = "yes"

    # restore the workflow
    restored_ctx = Context.from_dict(
        workflow, ctx_dict, serializer=JsonSerializer()
    )

    handler = workflow.run(ctx=restored_ctx)
    handler.ctx.send_event(
        HumanResponseEvent(
            response=response_str,
            user_name=input_ev.user_name,
        )
    )

    async for event in handler.stream_events():
        pass
        # print(type(event))
        # print("---")

    print("awaiting response", flush=True)
    response = await handler
    print("Got response: ", str(response), flush=True)
    return response


if __name__ == "__main__":
    asyncio.run(main())

Task exception was never retrieved
future: <Task finished name='_done' coro=<Workflow._start.<locals>._task() done, defined at /opt/homebrew/Caskroom/miniforge/base/envs/llamaindex-debug/lib/python3.13/site-packages/llama_index/core/workflow/workflow.py:258> exception=WorkflowDone()>
Traceback (most recent call last):
  File "/opt/homebrew/Caskroom/miniforge/base/envs/llamaindex-debug/lib/python3.13/site-packages/llama_index/core/workflow/workflow.py", line 303, in _task
    new_ev = await instrumented_step(**kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/miniforge/base/envs/llamaindex-debug/lib/python3.13/site-packages/llama_index/core/instrumentation/dispatcher.py", line 368, in async_wrapper
    result = await func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/miniforge/base/envs/llamaindex-debug/lib/python3.13/site-packages/llama_index/core/workflow/workflow.py", line 597, in _done
    raise WorkflowDone
llama_index.core.workflow.errors.WorkflowDone