feat: Add POST /events/{handler_id} endpoint and /handlers endpoint (#89)

* feat: Add POST /events/{handler_id} endpoint and /tasks endpoint

- Add GET /tasks endpoint to list running workflow tasks
- Add POST /events/{handler_id} endpoint to send events to running workflows
- Enable external event injection into workflow contexts
- Add InteractiveWorkflow test fixture for testing event handling
- Add comprehensive tests for both endpoints

This allows developers to:
1. Monitor running workflows via the /tasks endpoint
2. Send events to running workflows for interactive use cases
3. Build event-driven integrations with external systems

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

* fix lint

* refactor: Update /tasks endpoint to /handlers with enhanced status tracking

- Renamed /tasks endpoint to /handlers for better naming consistency
- Enhanced handler status tracking (running, completed, failed)
- Added result and error fields to handler response
- Fixed safe access to handler.result() and handler.exception() methods
- Added comprehensive tests for all handler states

The /handlers endpoint now provides complete visibility into workflow execution state.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

* update comments

---------

Co-authored-by: Claude <[email protected]>

Author: zhaotai

pyproject.toml

@@ -16,7 +16,7 @@ dev = [
 
 [project]
 name = "llama-index-workflows"
-version = "2.0.1"
+version = "2.1.0"
 description = "An event-driven, async-first, step-based way to control the execution flow of AI applications like Agents."
 readme = "README.md"
 license = "MIT"

src/workflows/server/server.py

@@ -69,11 +69,21 @@ def __init__(self, middleware: list[Middleware] | None = None):
                 self._stream_events,
                 methods=["GET"],
             ),
+            Route(
+                "/events/{handler_id}",
+                self._post_event,
+                methods=["POST"],
+            ),
             Route(
                 "/health",
                 self._health_check,
                 methods=["GET"],
             ),
+            Route(
+                "/handlers",
+                self._get_handlers,
+                methods=["GET"],
+            ),
         ]
 
         self.app = Starlette(routes=self._routes, middleware=self._middleware)
@@ -389,6 +399,156 @@ async def event_stream(handler: WorkflowHandler) -> AsyncGenerator[str, None]:
 
         return StreamingResponse(event_stream(handler), media_type=media_type)
 
+    async def _get_handlers(self, request: Request) -> JSONResponse:
+        """
+        ---
+        summary: Get handlers
+        description: Returns all workflow handlers.
+        responses:
+          200:
+            description: List of handlers
+            content:
+              application/json:
+                schema:
+                  type: object
+                  properties:
+                    handlers:
+                      type: array
+                      items:
+                        type: object
+                        properties:
+                          handler_id:
+                            type: string
+                          result:
+                            type: object
+                          error:
+                            type: object
+                          status:
+                            type: string
+                            enum: [running, completed, failed]
+                  required: [handlers]
+        """
+        handlers = []
+        for handler_id in self._handlers.keys():
+            handler = self._handlers[handler_id]
+            status = "running"
+            result = None
+            error = None
+
+            if handler.done():
+                try:
+                    result = handler.result()
+                    status = "completed"
+                except Exception as e:
+                    error = str(e)
+                    status = "failed"
+
+            handler_json = {
+                "handler_id": handler_id,
+                "status": status,
+                "result": result,
+                "error": error,
+            }
+            handlers.append(handler_json)
+
+        return JSONResponse({"handlers": handlers})
+
+    async def _post_event(self, request: Request) -> JSONResponse:
+        """
+        ---
+        summary: Send event to workflow
+        description: Sends an event to a running workflow's context.
+        parameters:
+          - in: path
+            name: handler_id
+            required: true
+            schema:
+              type: string
+            description: Workflow handler identifier.
+        requestBody:
+          required: true
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  event:
+                    type: string
+                    description: Serialized event in JSON format.
+                  step:
+                    type: string
+                    description: Optional target step name. If not provided, event is sent to all steps.
+                required: [event]
+        responses:
+          200:
+            description: Event sent successfully
+            content:
+              application/json:
+                schema:
+                  type: object
+                  properties:
+                    status:
+                      type: string
+                      enum: [sent]
+                  required: [status]
+          400:
+            description: Invalid event data
+          404:
+            description: Handler not found
+          409:
+            description: Workflow already completed
+        """
+        handler_id = request.path_params["handler_id"]
+
+        # Check if handler exists
+        handler = self._handlers.get(handler_id)
+        if handler is None:
+            raise HTTPException(detail="Handler not found", status_code=404)
+
+        # Check if workflow is still running
+        if handler.done():
+            raise HTTPException(detail="Workflow already completed", status_code=409)
+
+        # Get the context
+        ctx = handler.ctx
+        if ctx is None:
+            raise HTTPException(detail="Context not available", status_code=500)
+
+        # Parse request body
+        try:
+            body = await request.json()
+            event_str = body.get("event")
+            step = body.get("step")
+
+            if not event_str:
+                raise HTTPException(detail="Event data is required", status_code=400)
+
+            # Deserialize the event
+            serializer = JsonSerializer()
+            try:
+                event = serializer.deserialize(event_str)
+            except Exception as e:
+                raise HTTPException(
+                    detail=f"Failed to deserialize event: {e}", status_code=400
+                )
+
+            # Send the event to the context
+            try:
+                ctx.send_event(event, step=step)
+            except Exception as e:
+                raise HTTPException(
+                    detail=f"Failed to send event: {e}", status_code=400
+                )
+
+            return JSONResponse({"status": "sent"})
+
+        except HTTPException:
+            raise
+        except Exception as e:
+            raise HTTPException(
+                detail=f"Error processing request: {e}", status_code=500
+            )
+
     #
     # Private methods
     #

tests/server/conftest.py

@@ -42,6 +42,18 @@ async def stream_data(self, ctx: Context, ev: StartEvent) -> StopEvent:
         return StopEvent(result=f"completed_{count}_events")
 
 
+class ExternalEvent(Event):
+    message: str
+
+
+class InteractiveWorkflow(Workflow):
+    @step
+    async def start(self, ctx: Context, ev: StartEvent) -> StopEvent:
+        # Wait for an external event
+        external_event = await ctx.wait_for_event(ExternalEvent)
+        return StopEvent(result=f"received: {external_event.message}")
+
+
 @pytest.fixture
 def simple_test_workflow() -> Workflow:
     return SimpleTestWorkflow()
@@ -55,3 +67,8 @@ def error_workflow() -> Workflow:
 @pytest.fixture
 def streaming_workflow() -> Workflow:
     return StreamingWorkflow()
+
+
+@pytest.fixture
+def interactive_workflow() -> Workflow:
+    return InteractiveWorkflow()