diff --git a/.github/resources/integ-service-account.json.gpg b/.github/resources/integ-service-account.json.gpg
index 7740dccd8..5a52805c9 100644
Binary files a/.github/resources/integ-service-account.json.gpg and b/.github/resources/integ-service-account.json.gpg differ
diff --git a/.github/scripts/publish_preflight_check.sh b/.github/scripts/publish_preflight_check.sh
index 1d001c3b9..38fe49a88 100755
--- a/.github/scripts/publish_preflight_check.sh
+++ b/.github/scripts/publish_preflight_check.sh
@@ -159,8 +159,8 @@ echo_info "Generating changelog"
 echo_info "--------------------------------------------"
 echo_info ""
 
-echo_info "---< git fetch origin master --prune --unshallow >---"
-git fetch origin master --prune --unshallow
+echo_info "---< git fetch origin main --prune --unshallow >---"
+git fetch origin main --prune --unshallow
 echo ""
 
 echo_info "Generating changelog from history..."
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bfd29e2cc..5bf78a56b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -11,9 +11,20 @@ jobs:
         python: ['3.9', '3.10', '3.11', '3.12', '3.13', 'pypy3.9']
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
+
+    - name: Set up Python 3.13 for emulator
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
+      with:
+        python-version: '3.13'
+    - name: Setup functions emulator environment
+      run: |
+        python -m venv integration/emulators/functions/venv
+        source integration/emulators/functions/venv/bin/activate
+        pip install -r integration/emulators/functions/requirements.txt
+        deactivate
     - name: Set up Python ${{ matrix.python }}
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: ${{ matrix.python }}
     - name: Install dependencies
@@ -23,20 +34,27 @@ jobs:
     - name: Test with pytest
       run: pytest
     - name: Set up Node.js 20
-      uses: actions/setup-node@v4
+      uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # 4.4.0
       with:
         node-version: 20
-    - name: Run integration tests against emulator
-      run: |
-        npm install -g firebase-tools
-        firebase emulators:exec --only database --project fake-project-id 'pytest integration/test_db.py'
-
+    - name: Set up Java 21
+      uses: actions/setup-java@f2beeb24e141e01a676f977032f5a29d81c9e27e # 5.1.0
+      with:
+        distribution: 'temurin'
+        java-version: '21'
+        check-latest: true
+    - name: Install firebase-tools
+      run: npm install -g firebase-tools
+    - name: Run Database emulator tests
+      run: firebase emulators:exec --only database --project fake-project-id 'pytest integration/test_db.py'
+    - name: Run Functions emulator tests
+      run: firebase emulators:exec --config integration/emulators/firebase.json --only tasks,functions --project fake-project-id 'CLOUD_TASKS_EMULATOR_HOST=localhost:9499 pytest integration/test_functions.py'
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
     - name: Set up Python 3.9
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
     - name: Install dependencies
diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml
index 3d5420537..d60b3cd0b 100644
--- a/.github/workflows/nightly.yml
+++ b/.github/workflows/nightly.yml
@@ -29,12 +29,12 @@ jobs:
 
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@v4
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
       with:
         ref: ${{ github.event.client_payload.ref || github.ref }}
 
     - name: Set up Python
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
 
@@ -63,14 +63,14 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
       with:
         name: dist
         path: dist
 
     - name: Send email on failure
       if: failure()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@master
+      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
@@ -85,7 +85,7 @@ jobs:
 
     - name: Send email on cancelled
       if: cancelled()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@master
+      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 6cd1d3f07..53ebe825c 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -40,12 +40,12 @@ jobs:
     # via the 'ref' client parameter.
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@v4
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
       with:
         ref: ${{ github.event.client_payload.ref || github.ref }}
 
     - name: Set up Python
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
 
@@ -74,7 +74,7 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
       with:
         name: dist
         path: dist
@@ -84,15 +84,16 @@ jobs:
 
     # Check whether the release should be published. We publish only when the trigger PR is
     #   1. merged
-    #   2. to the master branch
+    #   2. to the main branch
     #   3. with the label 'release:publish', and
     #   4. the title prefix '[chore] Release '.
     if: github.event.pull_request.merged &&
-      github.ref == 'refs/heads/master' &&
+      github.ref == 'refs/heads/main' &&
       contains(github.event.pull_request.labels.*.name, 'release:publish') &&
       startsWith(github.event.pull_request.title, '[chore] Release ')
 
     runs-on: ubuntu-latest
+    environment: Release
     permissions:
       # Used to create a short-lived OIDC token which is given to PyPi to identify this workflow job
       # See: https://docs.github.com/en/actions/deployment/security-hardening-your-deployments/about-security-hardening-with-openid-connect#adding-permissions-settings
@@ -102,11 +103,11 @@ jobs:
 
     steps:
     - name: Checkout source for publish
-      uses: actions/checkout@v4
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
 
     # Download the artifacts created by the stage_release job.
     - name: Download release candidates
-      uses: actions/download-artifact@v4.1.7
+      uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
       with:
         name: dist
         path: dist
@@ -124,13 +125,13 @@ jobs:
             --notes '${{ steps.preflight.outputs.changelog }}'
 
     - name: Publish to Pypi
-      uses: pypa/gh-action-pypi-publish@release/v1
+      uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0
 
     # Post to Twitter if explicitly opted-in by adding the label 'release:tweet'.
     - name: Post to Twitter
       if: success() &&
         contains(github.event.pull_request.labels.*.name, 'release:tweet')
-      uses: firebase/firebase-admin-node/.github/actions/send-tweet@master
+      uses: firebase/firebase-admin-node/.github/actions/send-tweet@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
       with:
         status: >
           ${{ steps.preflight.outputs.version }} of @Firebase Admin Python SDK is available.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 72933a24f..139e7f96c 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -47,7 +47,7 @@ Great, we love hearing how we can improve our products! Share you idea through o
 ## <a name="submit"></a>Want to submit a pull request?
 
 Sweet, we'd love to accept your contribution!
-[Open a new pull request](https://github.com/firebase/firebase-admin-python/pull/new/master) and fill
+[Open a new pull request](https://github.com/firebase/firebase-admin-python/pull/new) and fill
 out the provided template.
 
 **If you want to implement a new feature, please open an issue with a proposal first so that we can
@@ -252,6 +252,17 @@ to ensure that exported user records contain the password hashes of the user acc
    3. Click **ADD ANOTHER ROLE** and choose **Firebase Authentication Admin**.
    4. Click **SAVE**.
 
+9. Enable Cloud Tasks:
+   1. Search for and enable **Cloud Run**.
+   2. Search for and enable **Cloud Tasks**.
+   3. Go to [Google Cloud console | IAM & admin](https://console.cloud.google.com/iam-admin)
+   and make sure your Firebase project is selected.
+   4. Ensure your service account has the following required roles:
+      * **Cloud Tasks Enqueuer** - `cloudtasks.taskEnqueuer`
+      * **Cloud Tasks Task Deleter** - `cloudtasks.taskDeleter`
+      * **Cloud Run Invoker** - `run.invoker`
+      * **Service Account User** - `iam.serviceAccountUser`
+
 
 Now you can invoke the integration test suite as follows:
 
diff --git a/firebase_admin/_auth_utils.py b/firebase_admin/_auth_utils.py
index a514442c4..8f3c419a7 100644
--- a/firebase_admin/_auth_utils.py
+++ b/firebase_admin/_auth_utils.py
@@ -479,7 +479,7 @@ def _parse_error_body(response):
         separator = code.find(':')
         if separator != -1:
             custom_message = code[separator + 1:].strip()
-            code = code[:separator]
+            code = code[:separator].strip()
 
     return code, custom_message
 
diff --git a/firebase_admin/credentials.py b/firebase_admin/credentials.py
index 7117b71a9..0edbecaae 100644
--- a/firebase_admin/credentials.py
+++ b/firebase_admin/credentials.py
@@ -37,7 +37,7 @@
 AccessTokenInfo = collections.namedtuple('AccessTokenInfo', ['access_token', 'expiry'])
 """Data included in an OAuth2 access token.
 
-Contains the access token string and the expiry time. The expirty time is exposed as a
+Contains the access token string and the expiry time. The expiry time is exposed as a
 ``datetime`` value.
 """
 
diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py
index 6db0fbb42..66ba700b3 100644
--- a/firebase_admin/functions.py
+++ b/firebase_admin/functions.py
@@ -18,11 +18,16 @@
 from datetime import datetime, timedelta, timezone
 from urllib import parse
 import re
+import os
 import json
 from base64 import b64encode
 from typing import Any, Optional, Dict
 from dataclasses import dataclass
+
 from google.auth.compute_engine import Credentials as ComputeEngineCredentials
+from google.auth.credentials import TokenState
+from google.auth.exceptions import RefreshError
+from google.auth.transport import requests as google_auth_requests
 
 import requests
 import firebase_admin
@@ -45,6 +50,8 @@
     'https://cloudtasks.googleapis.com/v2/' + _CLOUD_TASKS_API_RESOURCE_PATH
 _FIREBASE_FUNCTION_URL_FORMAT = \
     'https://{location_id}-{project_id}.cloudfunctions.net/{resource_id}'
+_EMULATOR_HOST_ENV_VAR = 'CLOUD_TASKS_EMULATOR_HOST'
+_EMULATED_SERVICE_ACCOUNT_DEFAULT = 'emulated-service-acct@email.com'
 
 _FUNCTIONS_HEADERS = {
     'X-GOOG-API-FORMAT-VERSION': '2',
@@ -54,6 +61,17 @@
 # Default canonical location ID of the task queue.
 _DEFAULT_LOCATION = 'us-central1'
 
+def _get_emulator_host() -> Optional[str]:
+    emulator_host = os.environ.get(_EMULATOR_HOST_ENV_VAR)
+    if emulator_host:
+        if '//' in emulator_host:
+            raise ValueError(
+                f'Invalid {_EMULATOR_HOST_ENV_VAR}: "{emulator_host}". It must follow format '
+                '"host:port".')
+        return emulator_host
+    return None
+
+
 def _get_functions_service(app) -> _FunctionsService:
     return _utils.get_app_service(app, _FUNCTIONS_ATTRIBUTE, _FunctionsService)
 
@@ -99,13 +117,19 @@ def __init__(self, app: App):
                 'projectId option, or use service account credentials. Alternatively, set the '
                 'GOOGLE_CLOUD_PROJECT environment variable.')
 
-        self._credential = app.credential.get_credential()
+        self._emulator_host = _get_emulator_host()
+        if self._emulator_host:
+            self._credential = _utils.EmulatorAdminCredentials()
+        else:
+            self._credential = app.credential.get_credential()
+
         self._http_client = _http_client.JsonHttpClient(credential=self._credential)
 
     def task_queue(self, function_name: str, extension_id: Optional[str] = None) -> TaskQueue:
         """Creates a TaskQueue instance."""
         return TaskQueue(
-            function_name, extension_id, self._project_id, self._credential, self._http_client)
+            function_name, extension_id, self._project_id, self._credential, self._http_client,
+            self._emulator_host)
 
     @classmethod
     def handle_functions_error(cls, error: Any):
@@ -121,7 +145,8 @@ def __init__(
             extension_id: Optional[str],
             project_id,
             credential,
-            http_client
+            http_client,
+            emulator_host: Optional[str] = None
         ) -> None:
 
         # Validate function_name
@@ -130,6 +155,7 @@ def __init__(
         self._project_id = project_id
         self._credential = credential
         self._http_client = http_client
+        self._emulator_host = emulator_host
         self._function_name = function_name
         self._extension_id = extension_id
         # Parse resources from function_name
@@ -163,16 +189,26 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str:
             str: The ID of the task relative to this queue.
         """
         task = self._validate_task_options(task_data, self._resource, opts)
-        service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT)
+        emulator_url = self._get_emulator_url(self._resource)
+        service_url = emulator_url or self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT)
         task_payload = self._update_task_payload(task, self._resource, self._extension_id)
         try:
             resp = self._http_client.body(
                 'post',
                 url=service_url,
                 headers=_FUNCTIONS_HEADERS,
-                json={'task': task_payload.__dict__}
+                json={'task': task_payload.to_api_dict()}
             )
-            task_name = resp.get('name', None)
+            if self._is_emulated():
+                # Emulator returns a response with format {task: {name: <task_name>}}
+                # The task name also has an extra '/' at the start compared to prod
+                task_info = resp.get('task') or {}
+                task_name = task_info.get('name')
+                if task_name:
+                    task_name = task_name[1:]
+            else:
+                # Production returns a response with format {name: <task_name>}
+                task_name = resp.get('name')
             task_resource = \
                 self._parse_resource_name(task_name, f'queues/{self._resource.resource_id}/tasks')
             return task_resource.resource_id
@@ -193,7 +229,11 @@ def delete(self, task_id: str) -> None:
             ValueError: If the input arguments are invalid.
         """
         _Validators.check_non_empty_string('task_id', task_id)
-        service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}')
+        emulator_url = self._get_emulator_url(self._resource)
+        if emulator_url:
+            service_url = emulator_url + f'/{task_id}'
+        else:
+            service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}')
         try:
             self._http_client.body(
                 'delete',
@@ -231,8 +271,8 @@ def _validate_task_options(
         """Validate and create a Task from optional ``TaskOptions``."""
         task_http_request = {
             'url': '',
-            'oidc_token': {
-                'service_account_email': ''
+            'oidcToken': {
+                'serviceAccountEmail': ''
             },
             'body': b64encode(json.dumps(data).encode()).decode(),
             'headers': {
@@ -246,7 +286,7 @@ def _validate_task_options(
                 task.http_request['headers'] = {**task.http_request['headers'], **opts.headers}
             if opts.schedule_time is not None and opts.schedule_delay_seconds is not None:
                 raise ValueError(
-                    'Both sechdule_delay_seconds and schedule_time cannot be set at the same time.')
+                    'Both schedule_delay_seconds and schedule_time cannot be set at the same time.')
             if opts.schedule_time is not None and opts.schedule_delay_seconds is None:
                 if not isinstance(opts.schedule_time, datetime):
                     raise ValueError('schedule_time should be UTC datetime.')
@@ -284,22 +324,53 @@ def _update_task_payload(self, task: Task, resource: Resource, extension_id: str
         """Prepares task to be sent with credentials."""
         # Get function url from task or generate from resources
         if not _Validators.is_non_empty_string(task.http_request['url']):
-            task.http_request['url'] = self._get_url(resource, _FIREBASE_FUNCTION_URL_FORMAT)
-        # If extension id is provided, it emplies that it is being run from a deployed extension.
+            if self._is_emulated():
+                task.http_request['url'] = ''
+            else:
+                task.http_request['url'] = self._get_url(resource, _FIREBASE_FUNCTION_URL_FORMAT)
+
+        # Refresh the credential to ensure all attributes (e.g. service_account_email, id_token)
+        # are populated, preventing cold start errors.
+        if self._credential.token_state != TokenState.FRESH:
+            try:
+                self._credential.refresh(google_auth_requests.Request())
+            except RefreshError as err:
+                raise ValueError(f'Initial task payload credential refresh failed: {err}') from err
+
+        # If extension id is provided, it implies that it is being run from a deployed extension.
         # Meaning that it's credential should be a Compute Engine Credential.
         if _Validators.is_non_empty_string(extension_id) and \
             isinstance(self._credential, ComputeEngineCredentials):
-
             id_token = self._credential.token
             task.http_request['headers'] = \
-                {**task.http_request['headers'], 'Authorization': f'Bearer ${id_token}'}
+                {**task.http_request['headers'], 'Authorization': f'Bearer {id_token}'}
             # Delete oidc token
-            del task.http_request['oidc_token']
+            del task.http_request['oidcToken']
         else:
-            task.http_request['oidc_token'] = \
-                {'service_account_email': self._credential.service_account_email}
+            try:
+                task.http_request['oidcToken'] = \
+                    {'serviceAccountEmail': self._credential.service_account_email}
+            except AttributeError as error:
+                if self._is_emulated():
+                    task.http_request['oidcToken'] = \
+                        {'serviceAccountEmail': _EMULATED_SERVICE_ACCOUNT_DEFAULT}
+                else:
+                    raise ValueError(
+                        'Failed to determine service account. Initialize the SDK with service '
+                        'account credentials or set service account ID as an app option.'
+                        ) from error
         return task
 
+    def _get_emulator_url(self, resource: Resource):
+        if self._emulator_host:
+            emulator_url_format = f'http://{self._emulator_host}/' + _CLOUD_TASKS_API_RESOURCE_PATH
+            url = self._get_url(resource, emulator_url_format)
+            return url
+        return None
+
+    def _is_emulated(self):
+        return self._emulator_host is not None
+
 
 class _Validators:
     """A collection of data validation utilities."""
@@ -424,6 +495,14 @@ class Task:
     schedule_time: Optional[str] = None
     dispatch_deadline: Optional[str] = None
 
+    def to_api_dict(self) -> dict:
+        """Converts the Task object to a dictionary suitable for the Cloud Tasks API."""
+        return {
+            'httpRequest': self.http_request,
+            'name': self.name,
+            'scheduleTime': self.schedule_time,
+            'dispatchDeadline': self.dispatch_deadline,
+        }
 
 @dataclass
 class Resource:
diff --git a/integration/emulators/.gitignore b/integration/emulators/.gitignore
new file mode 100644
index 000000000..b17f63107
--- /dev/null
+++ b/integration/emulators/.gitignore
@@ -0,0 +1,69 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+firebase-debug.log*
+firebase-debug.*.log*
+
+# Firebase cache
+.firebase/
+
+# Firebase config
+
+# Uncomment this if you'd like others to create their own Firebase project.
+# For a team working on the same Firebase project(s), it is recommended to leave
+# it commented so all members can deploy to the same project(s) in .firebaserc.
+# .firebaserc
+
+# Runtime data
+pids
+*.pid
+*.seed
+*.pid.lock
+
+# Directory for instrumented libs generated by jscoverage/JSCover
+lib-cov
+
+# Coverage directory used by tools like istanbul
+coverage
+
+# nyc test coverage
+.nyc_output
+
+# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files)
+.grunt
+
+# Bower dependency directory (https://bower.io/)
+bower_components
+
+# node-waf configuration
+.lock-wscript
+
+# Compiled binary addons (http://nodejs.org/api/addons.html)
+build/Release
+
+# Dependency directories
+node_modules/
+
+# Optional npm cache directory
+.npm
+
+# Optional eslint cache
+.eslintcache
+
+# Optional REPL history
+.node_repl_history
+
+# Output of 'npm pack'
+*.tgz
+
+# Yarn Integrity file
+.yarn-integrity
+
+# dotenv environment variables file
+.env
+
+# dataconnect generated files
+.dataconnect
diff --git a/integration/emulators/firebase.json b/integration/emulators/firebase.json
new file mode 100644
index 000000000..a7b727c4d
--- /dev/null
+++ b/integration/emulators/firebase.json
@@ -0,0 +1,29 @@
+{
+  "emulators": {
+    "tasks": {
+      "port": 9499
+    },
+    "ui": {
+      "enabled": false
+    },
+    "singleProjectMode": true,
+    "functions": {
+      "port": 5001
+    }
+  },
+  "functions": [
+    {
+      "source": "functions",
+      "codebase": "default",
+      "disallowLegacyRuntimeConfig": true,
+      "ignore": [
+        "venv",
+        ".git",
+        "firebase-debug.log",
+        "firebase-debug.*.log",
+        "*.local"
+      ],
+      "runtime": "python313"
+    }
+  ]
+}
diff --git a/integration/emulators/functions/.gitignore b/integration/emulators/functions/.gitignore
new file mode 100644
index 000000000..1609bab70
--- /dev/null
+++ b/integration/emulators/functions/.gitignore
@@ -0,0 +1,6 @@
+# Python bytecode
+__pycache__/
+
+# Python virtual environment
+venv/
+*.local
diff --git a/integration/emulators/functions/main.py b/integration/emulators/functions/main.py
new file mode 100644
index 000000000..6cd2c5766
--- /dev/null
+++ b/integration/emulators/functions/main.py
@@ -0,0 +1,7 @@
+from firebase_functions import tasks_fn
+
+@tasks_fn.on_task_dispatched()
+def testTaskQueue(req: tasks_fn.CallableRequest) -> None:
+    """Handles tasks from the task queue."""
+    print(f"Received task with data: {req.data}")
+    return
diff --git a/integration/emulators/functions/requirements.txt b/integration/emulators/functions/requirements.txt
new file mode 100644
index 000000000..6bbab42f8
--- /dev/null
+++ b/integration/emulators/functions/requirements.txt
@@ -0,0 +1 @@
+firebase_functions~=0.4.1
diff --git a/integration/test_auth.py b/integration/test_auth.py
index 7f4725dfe..b36063d19 100644
--- a/integration/test_auth.py
+++ b/integration/test_auth.py
@@ -724,6 +724,19 @@ def test_email_sign_in_with_settings(new_user_email_unverified, api_key):
     assert id_token is not None and len(id_token) > 0
     assert auth.get_user(new_user_email_unverified.uid).email_verified
 
+def test_auth_error_parse(new_user_email_unverified):
+    action_code_settings = auth.ActionCodeSettings(
+        ACTION_LINK_CONTINUE_URL, handle_code_in_app=True, link_domain="cool.link")
+    with pytest.raises(auth.InvalidHostingLinkDomainError) as excinfo:
+        auth.generate_sign_in_with_email_link(new_user_email_unverified.email,
+                                            action_code_settings=action_code_settings)
+    assert str(excinfo.value) == ('The provided hosting link domain is not configured in Firebase '
+                                  'Hosting or is not owned by the current project '
+                                  '(INVALID_HOSTING_LINK_DOMAIN). The provided hosting link '
+                                  'domain is not configured in Firebase Hosting or is not owned '
+                                  'by the current project. This cannot be a default hosting domain '
+                                  '(web.app or firebaseapp.com).')
+
 
 @pytest.fixture(scope='module')
 def oidc_provider():
diff --git a/integration/test_functions.py b/integration/test_functions.py
index 606798436..fc972f9e5 100644
--- a/integration/test_functions.py
+++ b/integration/test_functions.py
@@ -14,17 +14,34 @@
 
 """Integration tests for firebase_admin.functions module."""
 
+import os
 import pytest
 
 import firebase_admin
 from firebase_admin import functions
+from firebase_admin import _utils
 from integration import conftest
 
 
+_DEFAULT_DATA = {'data': {'city': 'Seattle'}}
+def integration_conf(request):
+    host_override = os.environ.get('CLOUD_TASKS_EMULATOR_HOST')
+    if host_override:
+        return _utils.EmulatorAdminCredentials(), 'fake-project-id'
+
+    return conftest.integration_conf(request)
+
 @pytest.fixture(scope='module')
 def app(request):
-    cred, _ = conftest.integration_conf(request)
-    return firebase_admin.initialize_app(cred, name='integration-functions')
+    cred, project_id = integration_conf(request)
+    return firebase_admin.initialize_app(
+        cred, options={'projectId': project_id}, name='integration-functions')
+
+@pytest.fixture(scope='module', autouse=True)
+def default_app():
+    # Overwrites the default_app fixture in conftest.py.
+    # This test suite should not use the default app. Use the app fixture instead.
+    pass
 
 
 class TestFunctions:
@@ -41,16 +58,31 @@ class TestFunctions:
     ]
 
     @pytest.mark.parametrize('task_queue_params', _TEST_FUNCTIONS_PARAMS)
-    def test_task_queue(self, task_queue_params):
-        queue = functions.task_queue(**task_queue_params)
-        assert queue is not None
-        assert callable(queue.enqueue)
-        assert callable(queue.delete)
-
-    @pytest.mark.parametrize('task_queue_params', _TEST_FUNCTIONS_PARAMS)
-    def test_task_queue_app(self, task_queue_params, app):
+    def test_task_queue(self, task_queue_params, app):
         assert app.name == 'integration-functions'
         queue = functions.task_queue(**task_queue_params, app=app)
         assert queue is not None
         assert callable(queue.enqueue)
         assert callable(queue.delete)
+
+    def test_task_enqueue(self, app):
+        queue = functions.task_queue('testTaskQueue', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert task_id is not None
+
+    @pytest.mark.skipif(
+        os.environ.get('CLOUD_TASKS_EMULATOR_HOST') is not None,
+        reason="Skipping test_task_delete against emulator due to bug in firebase-tools"
+    )
+    def test_task_delete(self, app):
+        # Skip this test against the emulator since tasks can't be delayed there to verify deletion
+        # See: https://github.com/firebase/firebase-tools/issues/8254
+        task_options = functions.TaskOptions(schedule_delay_seconds=60)
+        queue = functions.task_queue('testTaskQueue', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA, task_options)
+        assert task_id is not None
+        queue.delete(task_id)
+        # We don't have a way to check the contents of the queue so we check that the deleted
+        # task is not found using the delete method again.
+        with pytest.raises(firebase_admin.exceptions.NotFoundError):
+            queue.delete(task_id)
diff --git a/requirements.txt b/requirements.txt
index c68d71a0f..3b96eea00 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,5 @@
 astroid == 3.3.11
-pylint == 3.3.7
+pylint == 3.3.9
 pytest >= 8.2.2
 pytest-cov >= 2.4.0
 pytest-localserver >= 0.4.1
diff --git a/setup.cfg b/setup.cfg
index 32e00676b..4c6cf8d8f 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,4 +1,3 @@
 [tool:pytest]
 testpaths = tests
 asyncio_default_test_loop_scope = class
-asyncio_default_fixture_loop_scope = None
diff --git a/snippets/auth/index.py b/snippets/auth/index.py
index 6a509b8f5..656137dba 100644
--- a/snippets/auth/index.py
+++ b/snippets/auth/index.py
@@ -770,7 +770,7 @@ def get_tenant(tenant_id):
 	# [START get_tenant]
     tenant = tenant_mgt.get_tenant(tenant_id)
 
-    print('Retreieved tenant:', tenant.tenant_id)
+    print('Retrieved tenant:', tenant.tenant_id)
 	# [END get_tenant]
 
 def create_tenant():
diff --git a/tests/test_functions.py b/tests/test_functions.py
index 52e92c1b2..0f766767a 100644
--- a/tests/test_functions.py
+++ b/tests/test_functions.py
@@ -44,13 +44,14 @@ def setup_class(cls):
     def teardown_class(cls):
         testutils.cleanup_apps()
 
-    def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_RESPONSE):
+    def _instrument_functions_service(
+        self, app=None, status=200, payload=_DEFAULT_RESPONSE, mounted_url=_CLOUD_TASKS_URL):
         if not app:
             app = firebase_admin.get_app()
         functions_service = functions._get_functions_service(app)
         recorder = []
         functions_service._http_client.session.mount(
-            _CLOUD_TASKS_URL,
+            mounted_url,
             testutils.MockAdapter(payload, status, recorder))
         return functions_service, recorder
 
@@ -124,6 +125,10 @@ def test_task_enqueue(self):
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
         assert task_id == 'test-task-id'
 
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
     def test_task_enqueue_with_extension(self):
         resource_name = (
             'projects/test-project/locations/us-central1/queues/'
@@ -142,6 +147,59 @@ def test_task_enqueue_with_extension(self):
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
         assert task_id == 'test-task-id'
 
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
+    def test_task_enqueue_compute_engine(self):
+        app = firebase_admin.initialize_app(
+            testutils.MockComputeEngineCredential(),
+            options={'projectId': 'test-project'},
+            name='test-project-gce')
+        _, recorder = self._instrument_functions_service(app)
+        queue = functions.task_queue('test-function-name', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == _DEFAULT_REQUEST_URL
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+        assert recorder[0].headers['Authorization'] == 'Bearer mock-compute-engine-token'
+        expected_metrics_header = _utils.get_metrics_header() + ' mock-gce-cred-metric-tag'
+        assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
+        assert task_id == 'test-task-id'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-gce-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
+    def test_task_enqueue_with_extension_compute_engine(self):
+        resource_name = (
+            'projects/test-project/locations/us-central1/queues/'
+            'ext-test-extension-id-test-function-name/tasks'
+        )
+        extension_response = json.dumps({'name': resource_name + '/test-task-id'})
+        app = firebase_admin.initialize_app(
+            testutils.MockComputeEngineCredential(),
+            options={'projectId': 'test-project'},
+            name='test-project-gce-extensions')
+        _, recorder = self._instrument_functions_service(app, payload=extension_response)
+        queue = functions.task_queue('test-function-name', 'test-extension-id', app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == _CLOUD_TASKS_URL + resource_name
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+        assert recorder[0].headers['Authorization'] == 'Bearer mock-compute-engine-token'
+        expected_metrics_header = _utils.get_metrics_header() + ' mock-gce-cred-metric-tag'
+        assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
+        assert task_id == 'test-task-id'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert 'oidcToken' not in task['httpRequest']
+        assert task['httpRequest']['headers'] == {
+            'Content-Type': 'application/json',
+            'Authorization': 'Bearer mock-compute-engine-token'}
+
     def test_task_delete(self):
         _, recorder = self._instrument_functions_service()
         queue = functions.task_queue('test-function-name')
@@ -152,6 +210,58 @@ def test_task_delete(self):
         expected_metrics_header = _utils.get_metrics_header() + ' mock-cred-metric-tag'
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
 
+    def test_task_enqueue_with_emulator_host(self, monkeypatch):
+        emulator_host = 'localhost:8124'
+        emulator_url = f'http://{emulator_host}/'
+        request_url = emulator_url + _DEFAULT_TASK_PATH.replace('/tasks/test-task-id', '/tasks')
+
+        monkeypatch.setenv('CLOUD_TASKS_EMULATOR_HOST', emulator_host)
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(), {'projectId': 'test-project'}, name='emulator-app')
+
+        expected_task_name = (
+            '/projects/test-project/locations/us-central1'
+            '/queues/test-function-name/tasks/test-task-id'
+        )
+        expected_response = json.dumps({'task': {'name': expected_task_name}})
+        _, recorder = self._instrument_functions_service(
+            app, payload=expected_response, mounted_url=emulator_url)
+
+        queue = functions.task_queue('test-function-name', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == request_url
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {
+            'serviceAccountEmail': 'emulated-service-acct@email.com'
+        }
+        assert task_id == 'test-task-id'
+
+    def test_task_enqueue_without_emulator_host_error(self, monkeypatch):
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(),
+            {'projectId': 'test-project'}, name='no-emulator-app')
+
+        _, recorder = self._instrument_functions_service(app)
+        monkeypatch.delenv('CLOUD_TASKS_EMULATOR_HOST', raising=False)
+        queue = functions.task_queue('test-function-name', app=app)
+        with pytest.raises(ValueError) as excinfo:
+            queue.enqueue(_DEFAULT_DATA)
+        assert "Failed to determine service account" in str(excinfo.value)
+        assert len(recorder) == 0
+
+    def test_get_emulator_url_invalid_format(self, monkeypatch):
+        monkeypatch.setenv('CLOUD_TASKS_EMULATOR_HOST', 'http://localhost:8124')
+        app = firebase_admin.initialize_app(
+            testutils.MockCredential(), {'projectId': 'test-project'}, name='invalid-host-app')
+        with pytest.raises(ValueError) as excinfo:
+            functions.task_queue('test-function-name', app=app)
+        assert 'Invalid CLOUD_TASKS_EMULATOR_HOST' in str(excinfo.value)
+
 class TestTaskQueueOptions:
 
     _DEFAULT_TASK_OPTS = {'schedule_delay_seconds': None, 'schedule_time': None, \
@@ -202,13 +312,13 @@ def test_task_options_delay_seconds(self):
         assert len(recorder) == 1
         task = json.loads(recorder[0].body.decode())['task']
 
-        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        task_schedule_time = datetime.fromisoformat(task['scheduleTime'].replace('Z', '+00:00'))
         delta = abs(task_schedule_time - expected_schedule_time)
         assert delta <= timedelta(seconds=1)
 
-        assert task['dispatch_deadline'] == '200s'
-        assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
-        assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
+        assert task['dispatchDeadline'] == '200s'
+        assert task['httpRequest']['headers']['x-test-header'] == 'test-header-value'
+        assert task['httpRequest']['url'] in ['http://google.com', 'https://google.com']
         assert task['name'] == _DEFAULT_TASK_PATH
 
     def test_task_options_utc_time(self):
@@ -230,12 +340,12 @@ def test_task_options_utc_time(self):
         assert len(recorder) == 1
         task = json.loads(recorder[0].body.decode())['task']
 
-        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        task_schedule_time = datetime.fromisoformat(task['scheduleTime'].replace('Z', '+00:00'))
         assert task_schedule_time == expected_schedule_time
 
-        assert task['dispatch_deadline'] == '200s'
-        assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
-        assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
+        assert task['dispatchDeadline'] == '200s'
+        assert task['httpRequest']['headers']['x-test-header'] == 'test-header-value'
+        assert task['httpRequest']['url'] in ['http://google.com', 'https://google.com']
         assert task['name'] == _DEFAULT_TASK_PATH
 
     def test_schedule_set_twice_error(self):
@@ -247,7 +357,7 @@ def test_schedule_set_twice_error(self):
             queue.enqueue(_DEFAULT_DATA, opts)
         assert len(recorder) == 0
         assert str(excinfo.value) == \
-            'Both sechdule_delay_seconds and schedule_time cannot be set at the same time.'
+            'Both schedule_delay_seconds and schedule_time cannot be set at the same time.'
 
 
     @pytest.mark.parametrize('schedule_time', [
diff --git a/tests/testutils.py b/tests/testutils.py
index 598a929b4..7546595af 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -116,12 +116,25 @@ def __call__(self, *args, **kwargs): # pylint: disable=arguments-differ
 # pylint: disable=abstract-method
 class MockGoogleCredential(credentials.Credentials):
     """A mock Google authentication credential."""
+
+    def __init__(self):
+        super().__init__()
+        self.token = None
+        self._service_account_email = None
+        self._token_state = credentials.TokenState.INVALID
+
     def refresh(self, request):
         self.token = 'mock-token'
+        self._service_account_email = 'mock-email'
+        self._token_state = credentials.TokenState.FRESH
+
+    @property
+    def token_state(self):
+        return self._token_state
 
     @property
     def service_account_email(self):
-        return 'mock-email'
+        return self._service_account_email
 
     # Simulate x-goog-api-client modification in credential refresh
     def _metric_header_for_usage(self):
@@ -139,8 +152,24 @@ def get_credential(self):
 
 class MockGoogleComputeEngineCredential(compute_engine.Credentials):
     """A mock Compute Engine credential"""
+
+    def __init__(self):
+        super().__init__()
+        self.token = None
+        self._service_account_email = None
+        self._token_state = credentials.TokenState.INVALID
+
     def refresh(self, request):
         self.token = 'mock-compute-engine-token'
+        self._service_account_email = 'mock-gce-email'
+        self._token_state = credentials.TokenState.FRESH
+
+    @property
+    def token_state(self):
+        return self._token_state
+
+    def _metric_header_for_usage(self):
+        return 'mock-gce-cred-metric-tag'
 
 class MockComputeEngineCredential(firebase_admin.credentials.Base):
     """A mock Firebase credential implementation."""