diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py
index f8f50a2c8..de366792e 100644
--- a/firebase_admin/functions.py
+++ b/firebase_admin/functions.py
@@ -272,6 +272,11 @@ def _validate_task_options(
                         ', or underscores (_). The maximum length is 500 characters.')
                 task.name = self._get_url(
                     resource, _CLOUD_TASKS_API_RESOURCE_PATH + f'/{opts.task_id}')
+            if opts.uri is not None:
+                if not _Validators.is_url(opts.uri):
+                    raise ValueError(
+                        'uri must be a valid RFC3986 URI string using the https or http schema.')
+                task.http_request['url'] = opts.uri
         return task
 
     def _update_task_payload(self, task: Task, resource: Resource, extension_id: str) -> Task:
@@ -327,7 +332,7 @@ def is_url(cls, url: Any):
             return False
         try:
             parsed = parse.urlparse(url)
-            if not parsed.netloc:
+            if not parsed.netloc or parsed.scheme not in ['http', 'https']:
                 return False
             return True
         except Exception:   # pylint: disable=broad-except
@@ -382,12 +387,16 @@ class TaskOptions:
             By default, Content-Type is set to 'application/json'.
 
             The size of the headers must be less than 80KB.
+
+        uri: The full URL path that the request will be sent to. Must be a valid RFC3986 https or
+            http URL.
     """
     schedule_delay_seconds: Optional[int] = None
     schedule_time: Optional[datetime] = None
     dispatch_deadline_seconds: Optional[int] = None
     task_id: Optional[str] = None
     headers: Optional[Dict[str, str]] = None
+    uri: Optional[str] = None
 
 @dataclass
 class Task:
@@ -397,10 +406,10 @@ class Task:
     https://cloud.google.com/tasks/docs/reference/rest/v2/projects.locations.queues.tasks#resource:-task
 
     Args:
-        httpRequest:
-        name:
-        schedule_time:
-        dispatch_deadline:
+        httpRequest: The request to be made by the task worker.
+        name: The url path to identify the function.
+        schedule_time: The time when the task is scheduled to be attempted or retried.
+        dispatch_deadline: The deadline for requests sent to the worker.
     """
     http_request: Dict[str, Optional[str | dict]]
     name: Optional[str] = None
@@ -413,9 +422,9 @@ class Resource:
     """Contains the parsed address of a resource.
 
     Args:
-        resource_id:
-        project_id:
-        location_id:
+        resource_id: The ID of the resource.
+        project_id: The project ID of the resource.
+        location_id: The location ID of the resource.
     """
     resource_id: str
     project_id: Optional[str] = None
diff --git a/tests/test_functions.py b/tests/test_functions.py
index aecf26dae..75809c1ad 100644
--- a/tests/test_functions.py
+++ b/tests/test_functions.py
@@ -55,6 +55,13 @@ def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_R
             testutils.MockAdapter(payload, status, recorder))
         return functions_service, recorder
 
+    def test_task_queue_no_project_id(self):
+        def evaluate():
+            app = firebase_admin.initialize_app(testutils.MockCredential(), name='no-project-id')
+            with pytest.raises(ValueError):
+                functions.task_queue('test-function-name', app=app)
+        testutils.run_without_project_id(evaluate)
+
     @pytest.mark.parametrize('function_name', [
         'projects/test-project/locations/us-central1/functions/test-function-name',
         'locations/us-central1/functions/test-function-name',
@@ -179,14 +186,16 @@ def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_R
             'schedule_time': None,
             'dispatch_deadline_seconds': 200,
             'task_id': 'test-task-id',
-            'headers': {'x-test-header': 'test-header-value'}
+            'headers': {'x-test-header': 'test-header-value'},
+            'uri': 'https://google.com'
         },
         {
             'schedule_delay_seconds': None,
             'schedule_time': _SCHEDULE_TIME,
             'dispatch_deadline_seconds': 200,
             'task_id': 'test-task-id',
-            'headers': {'x-test-header': 'test-header-value'}
+            'headers': {'x-test-header': 'test-header-value'},
+            'uri': 'http://google.com'
         },
     ])
     def test_task_options(self, task_opts_params):
@@ -204,6 +213,7 @@ def test_task_options(self, task_opts_params):
 
         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['name'] == _DEFAULT_TASK_PATH
 
 
@@ -223,6 +233,7 @@ def test_schedule_set_twice_error(self):
         str(datetime.utcnow()),
         datetime.utcnow().isoformat(),
         datetime.utcnow().isoformat() + 'Z',
+        '', ' '
     ])
     def test_invalid_schedule_time_error(self, schedule_time):
         _, recorder = self._instrument_functions_service()
@@ -235,11 +246,7 @@ def test_invalid_schedule_time_error(self, schedule_time):
 
 
     @pytest.mark.parametrize('schedule_delay_seconds', [
-        -1,
-        '100',
-        '-1',
-        -1.23,
-        1.23
+        -1, '100', '-1', '', ' ', -1.23, 1.23
     ])
     def test_invalid_schedule_delay_seconds_error(self, schedule_delay_seconds):
         _, recorder = self._instrument_functions_service()
@@ -252,15 +259,7 @@ def test_invalid_schedule_delay_seconds_error(self, schedule_delay_seconds):
 
 
     @pytest.mark.parametrize('dispatch_deadline_seconds', [
-        14,
-        1801,
-        -15,
-        -1800,
-        0,
-        '100',
-        '-1',
-        -1.23,
-        1.23,
+        14, 1801, -15, -1800, 0, '100', '-1', '', ' ', -1.23, 1.23,
     ])
     def test_invalid_dispatch_deadline_seconds_error(self, dispatch_deadline_seconds):
         _, recorder = self._instrument_functions_service()
@@ -274,10 +273,7 @@ def test_invalid_dispatch_deadline_seconds_error(self, dispatch_deadline_seconds
 
 
     @pytest.mark.parametrize('task_id', [
-        'task/1',
-        'task.1',
-        'a'*501,
-        *non_alphanumeric_chars
+        '', ' ', 'task/1', 'task.1', 'a'*501, *non_alphanumeric_chars
     ])
     def test_invalid_task_id_error(self, task_id):
         _, recorder = self._instrument_functions_service()
@@ -290,3 +286,16 @@ def test_invalid_task_id_error(self, task_id):
             'task_id can contain only letters ([A-Za-z]), numbers ([0-9]), '
             'hyphens (-), or underscores (_). The maximum length is 500 characters.'
         )
+
+    @pytest.mark.parametrize('uri', [
+        '', ' ', 'a', 'foo', 'image.jpg', [], {}, True, 'google.com', 'www.google.com'
+    ])
+    def test_invalid_uri_error(self, uri):
+        _, recorder = self._instrument_functions_service()
+        opts = functions.TaskOptions(uri=uri)
+        queue = functions.task_queue('test-function-name')
+        with pytest.raises(ValueError) as excinfo:
+            queue.enqueue(_DEFAULT_DATA, opts)
+        assert len(recorder) == 0
+        assert str(excinfo.value) == \
+            'uri must be a valid RFC3986 URI string using the https or http schema.'
diff --git a/tests/testutils.py b/tests/testutils.py
index 0fb565106..ab4fb40cb 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -18,7 +18,7 @@
 
 import pytest
 
-from google.auth import credentials
+from google.auth import credentials, compute_engine
 from google.auth import transport
 from requests import adapters
 from requests import models
@@ -133,6 +133,19 @@ def __init__(self):
     def get_credential(self):
         return self._g_credential
 
+class MockGoogleComputeEngineCredential(compute_engine.Credentials):
+    """A mock Compute Engine credential"""
+    def refresh(self, request):
+        self.token = 'mock-compute-engine-token'
+
+class MockComputeEngineCredential(firebase_admin.credentials.Base):
+    """A mock Firebase credential implementation."""
+
+    def __init__(self):
+        self._g_credential = MockGoogleComputeEngineCredential()
+
+    def get_credential(self):
+        return self._g_credential
 
 class MockMultiRequestAdapter(adapters.HTTPAdapter):
     """A mock HTTP adapter that supports multiple responses for the Python requests module."""