Simplify schema for send event to handler in WorkflowServer (#129)

Author: logan-markewich

src/workflows/server/server.py

@@ -845,6 +845,8 @@ async def _post_event(self, request: Request) -> JSONResponse:
                   event:
                     type: string
                     description: Serialized event in JSON format.
+                    examples:
+                        {"type": "event_name", "data": {"key": "value"}}
                   step:
                     type: string
                     description: Optional target step name. If not provided, event is sent to all steps.
@@ -876,6 +878,10 @@ async def _post_event(self, request: Request) -> JSONResponse:
             raise HTTPException(detail="Handler not found", status_code=404)
 
         handler = wrapper.run_handler
+        events_by_title = {
+            e.__name__: e for e in self._workflows[wrapper.workflow_name].events
+        }
+
         # Check if workflow is still running
         if handler.done():
             raise HTTPException(detail="Workflow already completed", status_code=409)
@@ -897,7 +903,22 @@ async def _post_event(self, request: Request) -> JSONResponse:
             # Deserialize the event
             serializer = JsonSerializer()
             try:
-                event = serializer.deserialize(event_str)
+                event_data = serializer.deserialize(event_str)
+                if (
+                    isinstance(event_data, dict)
+                    and "type" in event_data
+                    and "data" in event_data
+                ):
+                    event_title = event_data["type"]
+                    event = events_by_title[event_title].model_validate(
+                        event_data["data"]
+                    )
+                elif isinstance(event_data, Event):
+                    event = event_data
+                else:
+                    raise ValueError(
+                        "Invalid event data. Should be a dictionary of {'type': 'event_type', 'data': {...}} or a serialized event"
+                    )
             except Exception as e:
                 raise HTTPException(
                     detail=f"Failed to deserialize event: {e}", status_code=400

tests/server/test_server_endpoints.py

@@ -711,6 +711,31 @@ async def test_post_event_to_running_workflow(client: AsyncClient) -> None:
     assert result["result"] == "received: Hello from test"
 
 
+@pytest.mark.asyncio
+async def test_post_event_simple_schema_to_running_workflow(
+    client: AsyncClient,
+) -> None:
+    # Start an interactive workflow
+    response = await client.post("/workflows/interactive/run-nowait", json={})
+    assert response.status_code == 200
+    handler_id = response.json()["handler_id"]
+
+    # Wait a bit for workflow to start
+    await asyncio.sleep(0.1)
+
+    # Send the event
+    event_str = '{"type": "ExternalEvent", "data": {"response": "Hello from test"}}'
+    response = await client.post(f"/events/{handler_id}", json={"event": event_str})
+    assert response.status_code == 200
+    assert response.json() == {"status": "sent"}
+
+    result = await wait_for_passing(
+        lambda: validate_result_response(handler_id, client)
+    )
+
+    assert result["result"] == "received: Hello from test"
+
+
 @pytest.mark.asyncio
 async def test_get_workflow_result_returns_202_when_pending(
     client: AsyncClient,
@@ -787,7 +812,9 @@ async def test_post_event_context_not_available(
     client: AsyncClient, server: WorkflowServer
 ) -> None:
     # Dumb test for code coverage. Inject a dummy handler with no context to trigger 500 path
-    wrapper = SimpleNamespace(run_handler=SimpleNamespace(done=lambda: False, ctx=None))
+    wrapper = SimpleNamespace(
+        run_handler=SimpleNamespace(done=lambda: False, ctx=None), workflow_name="test"
+    )
 
     handler_id = "noctx-1"
     server._handlers[handler_id] = wrapper  # type: ignore[assignment]