feat: add flag to run functions framework in asgi stack

.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}'). "

tests/conformance/async_main.py

@@ -0,0 +1,59 @@
+import asyncio
+import json
+
+from cloudevents.http import to_json
+
+import functions_framework.aio
+
+filename = "function_output.json"
+
+
+class RawJson:
+    data: dict
+
+    def __init__(self, data):
+        self.data = data
+
+    @staticmethod
+    def from_dict(obj: dict) -> "RawJson":
+        return RawJson(obj)
+
+    def to_dict(self) -> dict:
+        return self.data
+
+
+def _write_output(content):
+    with open(filename, "w") as f:
+        f.write(content)
+
+
+async def write_http(request):
+    json_data = await request.json()
+    _write_output(json.dumps(json_data))
+    return "OK", 200
+
+
+async def write_cloud_event(cloud_event):
+    _write_output(to_json(cloud_event).decode())
+
+
+@functions_framework.aio.http
+async def write_http_declarative(request):
+    json_data = await request.json()
+    _write_output(json.dumps(json_data))
+    return "OK", 200
+
+
+@functions_framework.aio.cloud_event
+async def write_cloud_event_declarative(cloud_event):
+    _write_output(to_json(cloud_event).decode())
+
+
+@functions_framework.aio.http
+async def write_http_declarative_concurrent(request):
+    await asyncio.sleep(1)
+    return "OK", 200
+
+
+# Note: Typed events are not supported in ASGI mode yet
+# Legacy event functions are also not supported in ASGI mode