feat: update hono API

README.md

@@ -157,6 +157,8 @@ Workflow can be used as middleware in any server framework, like `express`, `hon
 import { Hono } from "hono";
 import { serve } from "@hono/node-server";
 import { createHonoHandler } from "@llama-flow/core/interrupter/hono";
+import { filter } from "@llama-flow/core/stream/filter";
+import { until } from "@llama-flow/core/stream/until";
 import {
   agentWorkflow,
   startEvent,
@@ -168,9 +170,12 @@ const app = new Hono();
 app.post(
   "/workflow",
   createHonoHandler(
-    agentWorkflow,
-    async (ctx) => startEvent(await ctx.req.text()),
-    stopEvent,
+    toolCallWorkflow,
+    async (ctx, sendEvent) => {
+      sendEvent(startEvent.with(await ctx.req.text()));
+    },
+    (stream) =>
+      filter(until(stream, stopEvent), (event) => stopEvent.include(event)),
   ),
 );
 

demo/cloudflare/src/index.ts

@@ -2,6 +2,9 @@ import { Hono } from "hono";
 import { createWorkflow, workflowEvent } from "@llama-flow/core";
 import { createHonoHandler } from "@llama-flow/core/hono";
 import { html } from "hono/html";
+import { toolCallWorkflow } from "../../workflows/tool-call-agent";
+import { filter } from "@llama-flow/core/stream/filter";
+import { until } from "@llama-flow/core/stream/until";
 
 const app = new Hono();
 
@@ -16,9 +19,12 @@ workflow.handle([startEvent], ({ data }) => {
 app.post(
   "/workflow",
   createHonoHandler(
-    workflow,
-    async (ctx) => startEvent.with(await ctx.req.text()),
-    stopEvent,
+    toolCallWorkflow,
+    async (ctx, sendEvent) => {
+      sendEvent(startEvent.with(await ctx.req.text()));
+    },
+    (stream) =>
+      filter(until(stream, stopEvent), (event) => stopEvent.include(event)),
   ),
 );
 

demo/cloudflare/tsconfig.json

@@ -29,5 +29,5 @@
     ]
   },
   "exclude": ["node_modules", "dist", "tests"],
-  "include": ["src", "scripts"]
+  "include": ["src", "scripts", "../workflows"]
 }

demo/hono/app.ts

@@ -6,15 +6,20 @@ import {
   startEvent,
   stopEvent,
 } from "../workflows/tool-call-agent.js";
+import { until } from "@llama-flow/core/stream/until";
+import { filter } from "@llama-flow/core/stream/filter";
 
 const app = new Hono();
 
 app.post(
   "/workflow",
   createHonoHandler(
     toolCallWorkflow,
-    async (ctx) => startEvent.with(await ctx.req.text()),
-    stopEvent,
+    async (ctx, sendEvent) => {
+      sendEvent(startEvent.with(await ctx.req.text()));
+    },
+    (stream) =>
+      filter(until(stream, stopEvent), (event) => stopEvent.include(event)),
   ),
 );
 

docs/llamaflow/index.mdx

@@ -150,15 +150,20 @@ We provide tons of [middleware](#) and [integrations](#) to make it easy to ship
 import { Hono } from "hono";
 import { createHonoHandler } from "@llama-flow/core/interrupter/hono";
 import { openaiChatWorkflow, startEvent, stopEvent } from "@/lib/workflow";
+import { filter } from '@llama-flow/core/stream/filter'
+import { until } from '@llama-flow/core/stream/until'
 
 const app = new Hono();
 
 app.post(
   "/workflow",
   createHonoHandler(
-    openaiChatWorkflow,
-    async (ctx) => startEvent.with(ctx.req.json()),
-    stopEvent,
+    toolCallWorkflow,
+    async (ctx, sendEvent) => {
+      sendEvent(startEvent.with(await ctx.req.text()));
+    },
+    (stream) =>
+      filter(until(stream, stopEvent), (event) => stopEvent.include(event)),
   ),
 );
 

packages/core/src/hono.ts

@@ -1,26 +1,22 @@
 import type { Context, Handler } from "hono";
-import type {
-  Workflow,
-  WorkflowEvent,
-  WorkflowEventData,
+import {
+  type Workflow,
+  type WorkflowEventData,
+  WorkflowStream,
 } from "@llama-flow/core";
-import { runWorkflow } from "./stream/run";
 
 export const createHonoHandler = <Start, Stop>(
   workflow: Workflow,
-  getStart: (
+  initEvent: (
     c: Context,
-  ) => WorkflowEventData<Start> | Promise<WorkflowEventData<Start>>,
-  stopEvent: WorkflowEvent<Stop>,
-  wrapStopEvent?: (c: Context, stop: Stop) => Response,
+    sendEvent: (...events: WorkflowEventData<any>[]) => void,
+  ) => void | Promise<void>,
+  handleStream: (stream: WorkflowStream) => WorkflowStream,
 ): Handler => {
-  if (!wrapStopEvent) {
-    wrapStopEvent = (c, stop) => {
-      return c.json(stop as any);
-    };
-  }
-  return async (c) => {
-    const stop = await runWorkflow(workflow, await getStart(c), stopEvent);
-    return wrapStopEvent(c, stop.data);
+  return async (ctx) => {
+    const { stream, sendEvent } = workflow.createContext();
+    await initEvent(ctx, sendEvent);
+    const resultStream = handleStream(stream);
+    return resultStream.toResponse();
   };
 };

packages/core/src/stream/filter.ts

@@ -1,18 +1,11 @@
-import type {
-  WorkflowEvent,
-  WorkflowEventData,
-  WorkflowStream,
-} from "@llama-flow/core";
+import type { WorkflowEventData, WorkflowStream } from "@llama-flow/core";
 
-export function filter<
-  Event extends WorkflowEventData<any>,
-  Final extends Event,
->(
-  stream: ReadableStream<Event> | WorkflowStream<Event>,
-  cond: (event: Event) => event is Final,
-): ReadableStream<Final> | WorkflowStream<Final> {
+export function filter(
+  stream: WorkflowStream,
+  cond: (event: WorkflowEventData<any>) => boolean,
+): WorkflowStream {
   return stream.pipeThrough(
-    new TransformStream<Event, Final>({
+    new TransformStream({
       transform(event, controller) {
         if (cond(event)) {
           controller.enqueue(event);