chore(agents-api): Run sync system tool executions in new processes

agents-api/agents_api/activities/execute_system.py

@@ -18,12 +18,16 @@
 )
 from ..autogen.Sessions import CreateSessionRequest
 from ..autogen.Tools import SystemDef
-from ..common.protocol.remote import RemoteObject
 from ..common.protocol.tasks import StepContext
 from ..common.storage_handler import auto_blob_store, load_from_blob_store_if_remote
 from ..env import testing
 from ..models.developer import get_developer
 from .utils import get_handler
+from concurrent.futures import ProcessPoolExecutor
+from functools import partial
+
+# For running synchronous code in the background
+process_pool_executor = ProcessPoolExecutor()
 
 
 @auto_blob_store(deep=True)
@@ -108,13 +112,22 @@ async def execute_system(
             developer_id = arguments.pop("developer_id")
             session_id = arguments.pop("session_id", None)
             data = CreateSessionRequest(**arguments)
-            return handler(developer_id=developer_id, session_id=session_id, data=data)
+
+            # In case sessions.create becomes asynchronous in the future
+            if asyncio.iscoroutinefunction(handler):
+                return await handler()
+
+            # Run the synchronous function in another process
+            loop = asyncio.get_running_loop()
+            return await loop.run_in_executor(process_pool_executor, partial(handler, developer_id, session_id, data))
 
         # Handle regular operations
         if asyncio.iscoroutinefunction(handler):
             return await handler(**arguments)
-        return handler(**arguments)
-
+
+        # Run the synchronous function in another process
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(process_pool_executor, partial(handler, **arguments))
     except BaseException as e:
         if activity.in_activity():
             activity.logger.error(f"Error in execute_system_call: {e}")