sync plugin, DoTask function interface

Signed-off-by: Future Outlier <eric901201@gmai.com>

flytekit/extend/backend/agent_service.py

@@ -91,3 +91,28 @@ async def DeleteTask(self, request: DeleteTaskRequest, context: grpc.ServicerCon
             logger.error(f"failed to delete task with error {e}")
             context.set_code(grpc.StatusCode.INTERNAL)
             context.set_details(f"failed to delete task with error {e}")
+
+    async def DoTask(self, request: CreateTaskRequest, context: grpc.ServicerContext) -> GetTaskResponse:
+        try:
+            tmp = TaskTemplate.from_flyte_idl(request.template)
+            inputs = LiteralMap.from_flyte_idl(request.inputs) if request.inputs else None
+            agent = AgentRegistry.get_agent(context, tmp.type)
+            logger.info(f"{agent.task_type} agent start doing the job")
+            if agent.asynchronous:
+                try:
+                    return await agent.async_do(
+                        context=context, inputs=inputs, output_prefix=request.output_prefix, task_template=tmp
+                    )
+                except Exception as e:
+                    logger.error(f"failed to run async do with error {e}")
+                    raise e
+            try:
+                return await asyncio.to_thread(
+                    agent.do, context=context, inputs=inputs, output_prefix=request.output_prefix, task_template=tmp
+                )
+            except Exception as e:
+                logger
+        except Exception as e:
+            logger.error(f"failed to do task with error {e}")
+            context.set_code(grpc.StatusCode.INTERNAL)
+            context.set_details(f"failed to create task with error {e}")

flytekit/extend/backend/base_agent.py

@@ -79,6 +79,18 @@ def delete(self, context: grpc.ServicerContext, resource_meta: bytes) -> DeleteT
         """
         raise NotImplementedError
 
+    def do(
+        self,
+        context: grpc.ServicerContext,
+        output_prefix: str,
+        task_template: TaskTemplate,
+        inputs: typing.Optional[LiteralMap] = None,
+    ) -> CreateTaskResponse:
+        """
+        Return the result of executing a task. It should return error code if the task creation failed.
+        """
+        raise NotImplementedError
+
     async def async_create(
         self,
         context: grpc.ServicerContext,
@@ -105,6 +117,18 @@ async def async_delete(self, context: grpc.ServicerContext, resource_meta: bytes
         """
         raise NotImplementedError
 
+    def async_do(
+        self,
+        context: grpc.ServicerContext,
+        output_prefix: str,
+        task_template: TaskTemplate,
+        inputs: typing.Optional[LiteralMap] = None,
+    ) -> CreateTaskResponse:
+        """
+        Return the result of executing a task. It should return error code if the task creation failed.
+        """
+        raise NotImplementedError
+
 
 class AgentRegistry(object):
     """