feat: add flag to run functions framework in asgi stack (GoogleCloudPlatform#376)

* 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

* fix: apply black and isort formatting to source files

* test: add comprehensive tests for ASGI server support

- Add tests for HTTPServer ASGI/WSGI auto-detection
- Add tests for StarletteApplication and UvicornApplication
- Add tests for CLI --gateway flag functionality
- Add integration tests for async functions with ASGI
- Ensure 100% code coverage for new ASGI features
- Apply black and isort formatting to test files

* fix: skip async test files on Python 3.7

* fix: exclude async code from Python 3.7 coverage

* fix: Install uvicorn on windows.

* fix: unable to use reload in starlette.

* fix: add missing endpoint parameter to Route constructors in ASGI

* feat: add async conformance tests with ASGI gateway

* fix: set UvicornWorker class before parent init and update tests

* fix: update asgi tests to remove reload option

* feat: add async-specific conformance tests for ASGI mode

* fix: apply black formatting to async files

* fix: disable validateMapping for CloudEvent tests in ASGI mode

ASGI mode does not support automatic conversion from legacy events to CloudEvents, so validateMapping must be false for CloudEvent conformance tests.

* fix: avoid mutating options dict in Gunicorn applications

Create a copy of the options dict before modifying it to prevent side effects when the same options dict is reused elsewhere. This could cause issues with timeout tests.

* fix: add pragma comments for Python 3.7 coverage and fix options handling

- Add pragma: no cover comments for ASGI-specific code paths that won't execute in Python 3.7
- Fix options dict handling to use consistent variable names to avoid confusion

* fix: revert to separate GunicornApplication and UvicornApplication classes

Remove the BaseGunicornApplication abstraction as it was causing issues with the timeout mechanism. Each class now independently extends gunicorn.app.base.BaseApplication, which is cleaner and avoids the problems we were seeing with shared state and options handling.

* refactor: restore GunicornApplication to match main branch

Remove unnecessary changes to GunicornApplication class, keeping only
the UvicornApplication addition for ASGI support.

* chore: Untrack uv.lock

* chore: rename confirmance test (asgi) github workflow

* chore: cleanup .gitignore.

* chore: clean up unncessary comments.

Author: taeold

.coveragerc-py37

@@ -4,7 +4,18 @@
 # This file is only used by py37-* tox environments
 omit = 
     */functions_framework/aio/*
+    */functions_framework/_http/asgi.py
     */.tox/*
     */tests/*
     */venv/*
-    */.venv/*
+    */.venv/*
+
+[report]
+exclude_lines =
+    # Have to re-enable the standard pragma
+    pragma: no cover
+    
+    # Don't complain about async-specific imports and code
+    from functions_framework.aio import
+    from functions_framework._http.asgi import
+    from functions_framework._http.gunicorn import UvicornApplication

.github/workflows/conformance-asgi.yml

@@ -0,0 +1,91 @@
+name: Python Conformance CI (asgi)
+on:
+  push:
+    branches:
+      - 'main'
+  pull_request:
+
+# Declare default permissions as read only.
+permissions: read-all
+
+jobs:
+  build:
+    strategy:
+      matrix:
+        python: ['3.8', '3.9', '3.10', '3.11', '3.12']
+        platform: [ubuntu-latest]
+    runs-on: ${{ matrix.platform }}
+    steps:
+    - name: Harden Runner
+      uses: step-security/harden-runner@0634a2670c59f64b4a01f0f96f84700a4088b9f0 # v2.12.0
+      with:
+        disable-sudo: true
+        egress-policy: block
+        allowed-endpoints: >
+          api.github.com:443
+          files.pythonhosted.org:443
+          github.com:443
+          objects.githubusercontent.com:443
+          proxy.golang.org:443
+          pypi.org:443
+          storage.googleapis.com:443
+
+    - name: Checkout code
+      uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+
+    - name: Setup Python
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+      with:
+        python-version: ${{ matrix.python }}
+
+    - name: Install the framework with async extras
+      run: python -m pip install -e .[async]
+
+    - name: Setup Go
+      uses: actions/setup-go@d35c59abb061a4a6fb18e82ac0862c26744d6ab5 # v5.5.0
+      with:
+        go-version: '1.24'
+
+    - name: Run HTTP conformance tests
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@72a4f36b10f1c6435ab1a86a9ea24bda464cc262 # v1.8.6
+      with:
+        functionType: 'http'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/async_main.py --target write_http --signature-type http --gateway asgi'"
+
+    - name: Run CloudEvents conformance tests
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@72a4f36b10f1c6435ab1a86a9ea24bda464cc262 # v1.8.6
+      with:
+        functionType: 'cloudevent'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/async_main.py --target write_cloud_event --signature-type cloudevent --gateway asgi'"
+
+    - name: Run HTTP conformance tests declarative
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@72a4f36b10f1c6435ab1a86a9ea24bda464cc262 # v1.8.6
+      with:
+        functionType: 'http'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/async_main.py --target write_http_declarative --gateway asgi'"
+
+    - name: Run CloudEvents conformance tests declarative
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@72a4f36b10f1c6435ab1a86a9ea24bda464cc262 # v1.8.6
+      with:
+        functionType: 'cloudevent'
+        useBuildpacks: false
+        validateMapping: false
+        cmd: "'functions-framework --source tests/conformance/async_main.py --target write_cloud_event_declarative --gateway asgi'"
+
+    - name: Run HTTP concurrency tests declarative
+      uses: GoogleCloudPlatform/functions-framework-conformance/action@72a4f36b10f1c6435ab1a86a9ea24bda464cc262 # v1.8.6
+      with:
+        functionType: 'http'
+        useBuildpacks: false
+        validateConcurrency: true
+        cmd: "'functions-framework --source tests/conformance/async_main.py --target write_http_declarative_concurrent --gateway asgi'"
+
+    # Note: Event (legacy) and Typed tests are not supported in ASGI mode
+    # Note: validateMapping is set to false for CloudEvent tests because ASGI mode
+    # does not support automatic conversion from legacy events to CloudEvents

.gitignore

@@ -10,3 +10,4 @@ dist/
 function_output.json
 serverlog_stderr.txt
 serverlog_stdout.txt
+venv/

conftest.py

@@ -50,8 +50,8 @@ def pytest_ignore_collect(collection_path, config):
     if sys.version_info >= (3, 8):
         return None
 
-    # Skip test_aio.py entirely on Python 3.7
-    if collection_path.name == "test_aio.py":
+    # Skip test_aio.py and test_asgi.py entirely on Python 3.7
+    if collection_path.name in ["test_aio.py", "test_asgi.py"]:
         return True
 
     return None

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; python_version>='3.8'",
+    "uvicorn-worker>=0.2.0,<1.0.0; python_version>='3.8'"
+]
 
 [project.scripts]
 ff = "functions_framework._cli:_cli"

src/functions_framework/_cli.py

@@ -32,6 +32,19 @@
 @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":  # pragma: no cover
+        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
 
 
@@ -21,15 +23,30 @@ def __init__(self, app, debug, **options):
         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:
+        if isinstance(app, Flask):
+            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:  # pragma: no cover
+            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 +55,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,43 @@
+# 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
+
+        self.options = {
+            "log_level": "debug" if debug else "error",
+        }
+        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

@@ -70,3 +70,28 @@ 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(gunicorn.app.base.BaseApplication):
+    """Gunicorn application for ASGI apps using Uvicorn workers."""
+
+    def __init__(self, app, host, port, debug, **options):
+        self.options = {
+            "bind": "%s:%s" % (host, port),
+            "workers": int(os.environ.get("WORKERS", 1)),
+            "worker_class": "uvicorn_worker.UvicornWorker",
+            "timeout": int(os.environ.get("CLOUD_RUN_TIMEOUT_SECONDS", 0)),
+            "loglevel": os.environ.get("GUNICORN_LOG_LEVEL", "error"),
+            "limit_request_line": 0,
+        }
+        self.options.update(options)
+        self.app = app
+
+        super().__init__()
+
+    def load_config(self):
+        for key, value in self.options.items():
+            self.cfg.set(key, value)
+
+    def load(self):
+        return self.app

src/functions_framework/aio/__init__.py

@@ -197,14 +197,16 @@ def create_asgi_app(target=None, source=None, signature_type=None):
         routes.append(
             Route(
                 "/{path:path}",
-                http_handler,
+                endpoint=http_handler,
                 methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"],
             )
         )
     elif signature_type == _function_registry.CLOUDEVENT_SIGNATURE_TYPE:
         cloudevent_handler = _cloudevent_func_wrapper(function, is_async)
-        routes.append(Route("/{path:path}", cloudevent_handler, methods=["POST"]))
-        routes.append(Route("/", cloudevent_handler, methods=["POST"]))
+        routes.append(
+            Route("/{path:path}", endpoint=cloudevent_handler, methods=["POST"])
+        )
+        routes.append(Route("/", endpoint=cloudevent_handler, methods=["POST"]))
     elif signature_type == _function_registry.TYPED_SIGNATURE_TYPE:
         raise FunctionsFrameworkException(
             f"ASGI server does not support typed events (signature type: '{signature_type}'). "