Merge pull request #10 from thedadams/event-streaming

feat: add event streaming

Author: thedadams

README.md

@@ -181,6 +181,47 @@ print_output(out, err)
 wait()
 ```
 
+### `stream_exec_with_events(tool, opts)`
+
+This function streams the execution of a tool and returns the output, error, event stream, and process wait function. The streams must be read from.
+
+```python
+from gptscript.command import stream_exec_with_events
+from gptscript.tool import Tool
+
+tool = Tool(
+    json_response=True,
+    instructions="""
+Create three short graphic artist descriptions and their muses. 
+These should be descriptive and explain their point of view.
+Also come up with a made up name, they each should be from different
+backgrounds and approach art differently.
+the response should be in JSON and match the format:
+{
+   artists: [{
+      name: "name"
+      description: "description"
+   }]
+}
+""",
+    )
+
+def print_output(out, err, events):
+    for event in events:
+        print(event)
+ 
+    # Error stream has the debug info that is useful to see
+    for line in err:
+        print(line)
+
+    for line in out:
+        print(line)
+
+out, err, events, wait = stream_exec_with_events(tool)
+print_output(out, err, events)
+wait()
+```
+
 ### `stream_exec_file(tool_path, input="",opts)`
 
 This function streams the execution of a tool from a file and returns the output, error, and process wait function. The input values are passed to the tool as args.
@@ -201,6 +242,29 @@ print_output(out, err)
 wait()
 ```
 
+### `stream_exec_file_with_events(tool_path, input="",opts)`
+
+This function streams the execution of a tool from a file and returns the output, error, event stream, and process wait function. The input values are passed to the tool as args.
+
+```python
+from gptscript.command import stream_exec_file_with_events
+
+def print_output(out, err, events):
+    for event in events:
+        print(event)
+
+    # Error stream has the debug info that is useful to see
+    for line in err:
+        print(line)
+
+    for line in out:
+        print(line)
+
+out, err, events, wait = stream_exec_file_with_events("./init.gpt")
+print_output(out, err, events)
+wait()
+```
+
 ## Example Usage
 
 ```python

gptscript/command.py

@@ -1,7 +1,7 @@
 import os
 import sys
 
-from gptscript.exec_utils import exec_cmd, stream_exec_cmd
+from gptscript.exec_utils import exec_cmd, stream_exec_cmd, stream_exec_cmd_with_events
 from gptscript.tool import FreeForm, Tool
 
 optToArg = {
@@ -57,6 +57,18 @@ def stream_exec(tool, opts={}):
         raise e
 
 
+def stream_exec_with_events(tool, opts={}):
+    cmd = _get_command()
+    args = toArgs(opts)
+    args.append("-")
+    try:
+        tool_str = process_tools(tool)
+        process, events = stream_exec_cmd_with_events(cmd, args, input=tool_str)
+        return process.stdout, process.stderr, events, process.wait
+    except Exception as e:
+        raise e
+
+
 def exec_file(tool_path, input="", opts={}):
     cmd = _get_command()
     args = toArgs(opts)
@@ -88,6 +100,22 @@ def stream_exec_file(tool_path, input="", opts={}):
         raise e
 
 
+def stream_exec_file_with_events(tool_path, input="", opts={}):
+    cmd = _get_command()
+    args = toArgs(opts)
+
+    args.append(tool_path)
+
+    if input != "":
+        args.append(input)
+    try:
+        process, events = stream_exec_cmd_with_events(cmd, args)
+
+        return process.stdout, process.stderr, events, process.wait
+    except Exception as e:
+        raise e
+
+
 def toArgs(opts):
     args = ["--quiet=false"]
     for opt, val in opts.items():

gptscript/exec_utils.py

@@ -1,4 +1,11 @@
 import subprocess
+import os
+import sys
+
+if sys.platform == "win32":
+    import msvcrt
+    import win32api
+    import win32con
 
 
 def exec_cmd(cmd, args=[], input=None):
@@ -21,13 +28,55 @@ def exec_cmd(cmd, args=[], input=None):
 
 
 def stream_exec_cmd(cmd, args=[], input=None):
+    return _stream_cmd(cmd, args, input)
+
+
+def stream_exec_cmd_with_events(cmd, args=[], input=None):
+    events_out, events_in = os.pipe()
+    close_fds = True
+    fds = (events_in,)
+
+    if sys.platform == "win32":
+        # Can't close fds on Windows
+        close_fds = False
+        # Can't pass fds on Windows
+        fds = tuple()
+
+        # Convert the writable file descriptor to a handle
+        w_handle = msvcrt.get_osfhandle(events_in)
+
+        # Duplicate the handle to make it inheritable
+        proc_handle = win32api.GetCurrentProcess()
+        dup_handle = win32api.DuplicateHandle(
+            proc_handle,        # Source process handle
+            w_handle,           # Source handle
+            proc_handle,        # Target process handle
+            0,                  # Desired access (0 defaults to same as source)
+            1,                  # Inherit handle
+            win32con.DUPLICATE_SAME_ACCESS  # Options
+        )
+        args = ["--events-stream-to=fd://" + str(int(dup_handle))] + args
+    else:
+        args = ["--events-stream-to=fd://" + str(events_in)] + args
+
+    proc = _stream_cmd(cmd, args, input, fds=fds, close_fds=close_fds)
+    os.close(events_in)
+    if sys.platform == "win32":
+        win32api.CloseHandle(dup_handle)
+
+    return proc, os.fdopen(events_out, "r", encoding="utf-8")
+
+
+def _stream_cmd(cmd, args=[], input=None, fds=tuple(), close_fds=True):
     try:
         process = subprocess.Popen(
             [cmd] + args,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
+            pass_fds=fds,
+            close_fds=close_fds,
             encoding="utf-8",
         )
 

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "requests==2.31.0",
     "tqdm==4.66.2",
     "urllib3==2.2.1",
+    "pywin32==306 ; sys_platform == 'win32'",
 ]
 
 [project.urls]

requirements.txt

@@ -7,3 +7,4 @@ urllib3==2.2.1
 tox==4.14.1
 wheel==0.42.0
 setuptools==69.1.1
+pywin32==306; sys_platform == 'win32'

tests/test_gptscript.py

@@ -4,9 +4,10 @@
     list_models,
     list_tools,
     exec,
-    exec_file,
     stream_exec,
+    stream_exec_with_events,
     stream_exec_file,
+    stream_exec_file_with_events,
 )
 from gptscript.tool import Tool, FreeForm
 
@@ -115,7 +116,39 @@ def test_stream_exec_complex_tool(complex_tool):
 def test_stream_exec_file():
     out, err, wait = stream_exec_file("./fixtures/test.gpt")
     resp = wait()  # Wait for streaming to complete
+    for line in out:
+        print(line)
+    for line in err:
+        print(line)
     assert (
         out is not None or err is not None
     ), "Expected some output or error from stream_exec_file"
     assert resp == 0, "Expected a successful response from stream_exec_file"
+
+
+def test_stream_exec_tool_with_events(simple_tool):
+    out, err, events, wait = stream_exec_with_events(simple_tool)
+    has_events = False
+    for line in events:
+        has_events = line != ""
+
+    assert has_events, "Expected some events from stream_exec_with_events"
+    resp = wait()  # Wait for streaming to complete
+    assert (
+            out is not None or err is not None
+    ), "Expected some output or error from stream_exec_file"
+    assert resp == 0, "Expected a successful response from stream_exec_file"
+
+
+def test_stream_exec_file_with_events():
+    out, err, events, wait = stream_exec_file_with_events("./fixtures/test.gpt")
+    has_events = False
+    for line in events:
+        has_events = line != ""
+
+    assert has_events, "Expected some events from stream_exec_file_with_events"
+    resp = wait()  # Wait for streaming to complete
+    assert (
+            out is not None or err is not None
+    ), "Expected some output or error from stream_exec_file"
+    assert resp == 0, "Expected a successful response from stream_exec_file"