feat: add ASGI server support for async functions

- Add uvicorn and uvicorn-worker to async optional dependencies
- Refactor gunicorn.py with BaseGunicornApplication for shared config
- Add UvicornApplication class for ASGI apps
- Add StarletteApplication in asgi.py for development mode
- Update HTTPServer to auto-detect Flask (WSGI) vs other (ASGI) apps
- Add --gateway CLI flag to choose between wsgi and asgi
- Update test_http.py to use Flask instance in tests

Author: taeold

pyproject.toml

@@ -36,7 +36,11 @@ dependencies = [
 Homepage = "https://github.com/googlecloudplatform/functions-framework-python"
 
 [project.optional-dependencies]
-async = ["starlette>=0.37.0,<1.0.0; python_version>='3.8'"]
+async = [
+    "starlette>=0.37.0,<1.0.0; python_version>='3.8'",
+    "uvicorn>=0.18.0,<1.0.0; platform_system!='Windows' and python_version>='3.8'",
+    "uvicorn-worker>=0.2.0,<1.0.0; platform_system!='Windows' and python_version>='3.8'"
+]
 
 [project.scripts]
 ff = "functions_framework._cli:_cli"

src/functions_framework/_cli.py

@@ -32,6 +32,18 @@
 @click.option("--host", envvar="HOST", type=click.STRING, default="0.0.0.0")
 @click.option("--port", envvar="PORT", type=click.INT, default=8080)
 @click.option("--debug", envvar="DEBUG", is_flag=True)
-def _cli(target, source, signature_type, host, port, debug):
-    app = create_app(target, source, signature_type)
+@click.option(
+    "--gateway",
+    envvar="GATEWAY",
+    type=click.Choice(["wsgi", "asgi"]),
+    default="wsgi",
+    help="Server gateway interface type (wsgi for sync, asgi for async)"
+)
+def _cli(target, source, signature_type, host, port, debug, gateway):
+    if gateway == "asgi":
+        from functions_framework.aio import create_asgi_app
+        app = create_asgi_app(target, source, signature_type)
+    else:
+        app = create_app(target, source, signature_type)
+    
     create_server(app, debug).run(host, port)

src/functions_framework/_http/__init__.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from flask import Flask
+
 from functions_framework._http.flask import FlaskApplication
 
 
@@ -20,16 +22,30 @@ def __init__(self, app, debug, **options):
         self.app = app
         self.debug = debug
         self.options = options
-
-        if self.debug:
-            self.server_class = FlaskApplication
-        else:
-            try:
-                from functions_framework._http.gunicorn import GunicornApplication
-
-                self.server_class = GunicornApplication
-            except ImportError as e:
+        
+        # Check if app is Flask (WSGI) or not (ASGI)
+        if isinstance(app, Flask):
+            # WSGI app
+            if self.debug:
                 self.server_class = FlaskApplication
+            else:
+                try:
+                    from functions_framework._http.gunicorn import GunicornApplication
+                    self.server_class = GunicornApplication
+                except ImportError as e:
+                    self.server_class = FlaskApplication
+        else:
+            # ASGI app (Starlette or other)
+            if self.debug:
+                from functions_framework._http.asgi import StarletteApplication
+                self.server_class = StarletteApplication
+            else:
+                try:
+                    from functions_framework._http.gunicorn import UvicornApplication
+                    self.server_class = UvicornApplication
+                except ImportError as e:
+                    from functions_framework._http.asgi import StarletteApplication
+                    self.server_class = StarletteApplication
 
     def run(self, host, port):
         http_server = self.server_class(
@@ -38,5 +54,5 @@ def run(self, host, port):
         http_server.run()
 
 
-def create_server(wsgi_app, debug, **options):
-    return HTTPServer(wsgi_app, debug, **options)
+def create_server(app, debug, **options):
+    return HTTPServer(app, debug, **options)

src/functions_framework/_http/asgi.py

@@ -0,0 +1,50 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import uvicorn
+
+
+class StarletteApplication:
+    """A Starlette application that uses Uvicorn for direct serving (development mode)."""
+    
+    def __init__(self, app, host, port, debug, **options):
+        """Initialize the Starlette application.
+        
+        Args:
+            app: The ASGI application to serve
+            host: The host to bind to
+            port: The port to bind to
+            debug: Whether to run in debug mode
+            **options: Additional options to pass to Uvicorn
+        """
+        self.app = app
+        self.host = host
+        self.port = port
+        self.debug = debug
+        
+        # Default uvicorn config
+        self.options = {
+            "log_level": "debug" if debug else "error",
+            "reload": debug,
+        }
+        self.options.update(options)
+    
+    def run(self):
+        """Run the Uvicorn server directly."""
+        uvicorn.run(
+            self.app,
+            host=self.host,
+            port=int(self.port),
+            **self.options
+        )

src/functions_framework/_http/gunicorn.py

@@ -27,32 +27,20 @@
 TIMEOUT_SECONDS = None
 
 
-class GunicornApplication(gunicorn.app.base.BaseApplication):
+class BaseGunicornApplication(gunicorn.app.base.BaseApplication):
+    """Base Gunicorn application with common configuration."""
     def __init__(self, app, host, port, debug, **options):
-        threads = int(os.environ.get("THREADS", (os.cpu_count() or 1) * 4))
-
         global TIMEOUT_SECONDS
         TIMEOUT_SECONDS = int(os.environ.get("CLOUD_RUN_TIMEOUT_SECONDS", 0))
 
         self.options = {
             "bind": "%s:%s" % (host, port),
             "workers": int(os.environ.get("WORKERS", 1)),
-            "threads": threads,
             "loglevel": os.environ.get("GUNICORN_LOG_LEVEL", "error"),
             "limit_request_line": 0,
+            "timeout": TIMEOUT_SECONDS,
         }
 
-        if (
-            TIMEOUT_SECONDS > 0
-            and threads > 1
-            and (os.environ.get("THREADED_TIMEOUT_ENABLED", "False").lower() == "true")
-        ):  # pragma: no cover
-            self.options["worker_class"] = (
-                "functions_framework._http.gunicorn.GThreadWorkerWithTimeoutSupport"
-            )
-        else:
-            self.options["timeout"] = TIMEOUT_SECONDS
-
         self.options.update(options)
         self.app = app
 
@@ -66,7 +54,36 @@ def load(self):
         return self.app
 
 
+class GunicornApplication(BaseGunicornApplication):
+    """Gunicorn application for WSGI apps with gthread worker support."""
+    def __init__(self, app, host, port, debug, **options):
+        threads = int(os.environ.get("THREADS", (os.cpu_count() or 1) * 4))
+        options["threads"] = threads
+        
+        super().__init__(app, host, port, debug, **options)
+        
+        # Use custom worker with timeout support if conditions are met
+        if (
+            TIMEOUT_SECONDS > 0
+            and threads > 1
+            and (os.environ.get("THREADED_TIMEOUT_ENABLED", "False").lower() == "true")
+        ):  # pragma: no cover
+            self.options["worker_class"] = (
+                "functions_framework._http.gunicorn.GThreadWorkerWithTimeoutSupport"
+            )
+            # Remove timeout from options when using custom worker
+            del self.options["timeout"]
+
+
 class GThreadWorkerWithTimeoutSupport(ThreadWorker):  # pragma: no cover
     def handle_request(self, req, conn):
         with ThreadingTimeout(TIMEOUT_SECONDS):
             super(GThreadWorkerWithTimeoutSupport, self).handle_request(req, conn)
+
+
+class UvicornApplication(BaseGunicornApplication):
+    """Gunicorn application for ASGI apps using Uvicorn workers."""
+    def __init__(self, app, host, port, debug, **options):
+        super().__init__(app, host, port, debug, **options)
+        
+        self.options["worker_class"] = "uvicorn_worker.UvicornWorker"

tests/test_http.py

@@ -16,6 +16,7 @@
 import platform
 import sys
 
+import flask
 import pretend
 import pytest
 
@@ -45,7 +46,8 @@ def test_create_server(monkeypatch, debug):
     ],
 )
 def test_httpserver(monkeypatch, debug, gunicorn_missing, expected):
-    app = pretend.stub()
+    # Create a mock Flask app
+    app = flask.Flask("test")
     http_server = pretend.stub(run=pretend.call_recorder(lambda: None))
     server_classes = {
         "flask": pretend.call_recorder(lambda *a, **kw: http_server),