update agentapp run method (#161)

Author: zzhangpurdue

pyproject.toml

@@ -25,6 +25,10 @@ dependencies = [
 
 [tool.setuptools]
 packages = { find = { where = ["src"] } }
+include-package-data = true
+
+[tool.setuptools.package-data]
+"agentscope_runtime.engine.deployers.utils.service_utils" = ["*.j2"]
 
 [build-system]
 requires = ["setuptools>=42", "wheel"]

src/agentscope_runtime/engine/app/agent_app.py

@@ -4,6 +4,7 @@
 from contextlib import asynccontextmanager
 from typing import Optional, Any, Callable, List
 
+import uvicorn
 from fastapi import FastAPI
 from pydantic import BaseModel
 
@@ -13,6 +14,11 @@
 from ..deployers.adapter.responses.response_api_protocol_adapter import (
     ResponseAPIDefaultAdapter,
 )
+from ..deployers.utils.deployment_modes import DeploymentMode
+from ..deployers.utils.service_utils.fastapi_factory import FastAPIAppFactory
+from ..deployers.utils.service_utils.service_config import (
+    DEFAULT_SERVICES_CONFIG,
+)
 from ..runner import Runner
 from ..schemas.agent_schemas import AgentRequest
 from ..services.context_manager import ContextManager
@@ -57,6 +63,8 @@ def __init__(
         self.request_model = request_model
         self.before_start = before_start
         self.after_finish = after_finish
+        self.broker_url = broker_url
+        self.backend_url = backend_url
 
         self._agent = agent
         self._runner = None
@@ -116,39 +124,62 @@ def run(
         host="0.0.0.0",
         port=8090,
         embed_task_processor=False,
+        services_config=None,
         **kwargs,
     ):
+        """
+        Run the AgentApp using FastAPIAppFactory directly.
+
+        Args:
+            host: Host to bind to
+            port: Port to bind to
+            embed_task_processor: Whether to embed task processor
+            services_config: Optional services configuration
+            **kwargs: Additional keyword arguments
+        """
+
         try:
-            loop = asyncio.get_event_loop()
-            if self._runner is not None:
-                loop.run_until_complete(self._runner.__aenter__())
+            logger.info(
+                "[AgentApp] Starting AgentApp with FastAPIAppFactory...",
+            )
+
+            # Use default services config if not provided
+            if services_config is None:
+                services_config = DEFAULT_SERVICES_CONFIG
+
+            # Create FastAPI application using the factory
+            fastapi_app = FastAPIAppFactory.create_app(
+                runner=self._runner,
+                endpoint_path=self.endpoint_path,
+                request_model=self.request_model,
+                response_type=self.response_type,
+                stream=self.stream,
+                before_start=self.before_start,
+                after_finish=self.after_finish,
+                mode=DeploymentMode.DAEMON_THREAD,
+                services_config=services_config,
+                protocol_adapters=self.protocol_adapters,
+                custom_endpoints=self.custom_endpoints,
+                broker_url=self.broker_url,
+                backend_url=self.backend_url,
+                enable_embedded_worker=embed_task_processor,
+                **kwargs,
+            )
 
-            logger.info("[AgentApp] Runner initialized.")
+            logger.info(f"[AgentApp] Starting server on {host}:{port}")
 
-            super().run(
+            # Start the FastAPI application with uvicorn
+            uvicorn.run(
+                fastapi_app,
                 host=host,
                 port=port,
-                embed_task_processor=embed_task_processor,
-                **kwargs,
+                log_level="info",
+                access_log=True,
             )
 
         except Exception as e:
             logger.error(f"[AgentApp] Error while running: {e}")
-
-        finally:
-            try:
-                try:
-                    loop = asyncio.get_event_loop()
-                except RuntimeError:
-                    loop = asyncio.new_event_loop()
-                    asyncio.set_event_loop(loop)
-                if self._runner is not None:
-                    loop.run_until_complete(
-                        self._runner.__aexit__(None, None, None),
-                    )
-                    logger.info("[AgentApp] Runner cleaned up.")
-            except Exception as e:
-                logger.error(f"[AgentApp] Error while cleaning up runner: {e}")
+            raise
 
     async def deploy(self, deployer, **kwargs):
         """Deploy the agent app with custom endpoints support"""

src/agentscope_runtime/engine/deployers/base.py

@@ -9,6 +9,7 @@
 class DeployManager(ABC):
     def __init__(self):
         self.deploy_id = str(uuid.uuid4())
+        self._app = None
 
     @abstractmethod
     async def deploy(self, *args, **kwargs) -> Dict[str, str]:

src/agentscope_runtime/engine/deployers/local_deployer.py

@@ -29,7 +29,8 @@ def __init__(
         self,
         host: str = "127.0.0.1",
         port: int = 8000,
-        shutdown_timeout: int = 120,
+        shutdown_timeout: int = 30,
+        startup_timeout: int = 30,
         logger: Optional[logging.Logger] = None,
     ):
         """Initialize LocalDeployManager.
@@ -44,6 +45,7 @@ def __init__(
         self.host = host
         self.port = port
         self._shutdown_timeout = shutdown_timeout
+        self._startup_timeout = startup_timeout
         self._logger = logger or logging.getLogger(__name__)
 
         # State management
@@ -66,6 +68,7 @@ def __init__(
 
     async def deploy(
         self,
+        app=None,
         runner: Optional[Any] = None,
         endpoint_path: str = "/process",
         request_model: Optional[Type] = None,
@@ -111,6 +114,19 @@ async def deploy(
         if self.is_running:
             raise RuntimeError("Service is already running")
 
+        self._app = app
+        if self._app is not None:
+            runner = self._app._runner
+            endpoint_path = self._app.endpoint_path
+            response_type = self._app.response_type
+            stream = self._app.stream
+            request_model = self._app.request_model
+            before_start = self._app.before_start
+            after_finish = self._app.after_finish
+            backend_url = self._app.backend_url
+            broker_url = self._app.broker_url
+            custom_endpoints = self._app.custom_endpoints
+            protocol_adapters = self._app.protocol_adapters
         try:
             if mode == DeploymentMode.DAEMON_THREAD:
                 return await self._deploy_daemon_thread(
@@ -194,7 +210,7 @@ def run_server():
         self._server_thread.start()
 
         # Wait for server to start
-        await self._wait_for_server_ready()
+        await self._wait_for_server_ready(self._startup_timeout)
 
         self.is_running = True
         self.deploy_id = f"daemon_{self.host}_{self.port}"

src/agentscope_runtime/engine/deployers/modelstudio_deployer.py

@@ -201,16 +201,6 @@ async def _oss_create_bucket_if_not_exists(client, bucket_name: str) -> None:
         )
 
 
-def _create_bucket_name(prefix: str, base_name: str) -> str:
-    import re as _re
-
-    ts = time.strftime("%Y%m%d-%H%M%S", time.gmtime())
-    base = _re.sub(r"\s+", "-", base_name)
-    base = _re.sub(r"[^a-zA-Z0-9-]", "", base).lower().strip("-")
-    name = f"{prefix}-{base}-{ts}"
-    return name[:63]
-
-
 async def _oss_put_and_presign(
     client,
     bucket_name: str,

src/agentscope_runtime/engine/deployers/utils/service_utils/fastapi_factory.py

@@ -274,9 +274,10 @@ async def _handle_shutdown(
             and not app.state.runner_managed_externally
         ):
             runner = app.state.runner
-            if runner and hasattr(runner, "context_manager"):
+            if runner:
                 try:
-                    await runner.context_manager.__aexit__(None, None, None)
+                    # Clean up runner
+                    await runner.__aexit__(None, None, None)
                 except Exception as e:
                     print(f"Warning: Error during runner cleanup: {e}")
 
@@ -297,15 +298,15 @@ async def _create_internal_runner(services_config: ServicesConfig):
             memory_service=services["memory"],
         )
 
-        # Initialize context manager
-        await context_manager.__aenter__()
-
         # Create runner (agent will be set later)
         runner = Runner(
             agent=None,  # Will be set by the specific deployment
             context_manager=context_manager,
         )
 
+        # Initialize runner
+        await runner.__aenter__()
+
         return runner
 
     @staticmethod
@@ -673,18 +674,48 @@ def _create_streaming_parameter_wrapper(
         try:
             sig = inspect.signature(handler)
             params = list(sig.parameters.values())
+            no_params = False
+            param_annotation = None
 
             if not params:
-                # No parameters, call function directly
-                return handler
+                no_params = True
+            else:
+                # Get the first parameter
+                first_param = params[0]
+                param_annotation = first_param.annotation
 
-            # Get the first parameter
-            first_param = params[0]
-            param_annotation = first_param.annotation
+                # If no annotation or annotation is Request, goto no params
+                # logic
+                if param_annotation in [inspect.Parameter.empty, Request]:
+                    no_params = True
 
-            # If no annotation or annotation is Request, pass Request directly
-            if param_annotation in [inspect.Parameter.empty, Request]:
-                return handler
+            if no_params:
+                if is_async_gen:
+
+                    async def async_no_param_wrapper():
+                        async def generate():
+                            async for chunk in handler():
+                                yield str(chunk)
+
+                        return StreamingResponse(
+                            generate(),
+                            media_type="text/plain",
+                        )
+
+                    return async_no_param_wrapper
+                else:
+
+                    async def sync_no_param_wrapper():
+                        def generate():
+                            for chunk in handler():
+                                yield str(chunk)
+
+                        return StreamingResponse(
+                            generate(),
+                            media_type="text/plain",
+                        )
+
+                    return sync_no_param_wrapper
 
             # Check if the annotation is a Pydantic model
             if isinstance(param_annotation, type) and issubclass(
@@ -693,24 +724,20 @@ def _create_streaming_parameter_wrapper(
             ):
                 if is_async_gen:
 
-                    async def async_stream_pydantic_wrapper(request: Request):
+                    async def async_stream_pydantic_wrapper(
+                        request: Request,
+                    ):
                         try:
                             body = await request.json()
                             parsed_param = param_annotation(**body)
 
-                            # Create async generator and return
-                            # StreamingResponse
                             async def generate():
                                 async for chunk in handler(parsed_param):
                                     yield str(chunk)
 
                             return StreamingResponse(
                                 generate(),
-                                media_type="text/event-stream",
-                                headers={
-                                    "Cache-Control": "no-cache",
-                                    "Connection": "keep-alive",
-                                },
+                                media_type="text/plain",
                             )
                         except Exception as e:
                             return StreamingResponse(
@@ -721,7 +748,9 @@ async def generate():
                     return async_stream_pydantic_wrapper
                 else:
 
-                    async def sync_stream_pydantic_wrapper(request: Request):
+                    async def sync_stream_pydantic_wrapper(
+                        request: Request,
+                    ):
                         try:
                             body = await request.json()
                             parsed_param = param_annotation(**body)
@@ -738,19 +767,16 @@ def generate():
                             return JSONResponse(
                                 status_code=422,
                                 content={
-                                    "detail": f"Request parsing error: "
-                                    f"{str(e)}",
+                                    "detail": f"Request parsing error:"
+                                    f" {str(e)}",
                                 },
                             )
 
                     return sync_stream_pydantic_wrapper
 
-            # For other types, fall back to original behavior
             return handler
 
         except Exception:
-            # If anything goes wrong with introspection, fall back to
-            # original behavior
             return handler
 
     @staticmethod