diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py
index 7eb3072fa..6ba729008 100644
--- a/src/a2a/utils/push_url_validator.py
+++ b/src/a2a/utils/push_url_validator.py
@@ -1,29 +1,15 @@
-"""Shared policy for screening client-supplied push-notification URLs."""
+"""Backward-compatible entry point for push-notification URL screening.
 
-import asyncio
-import ipaddress
-import logging
-import socket
-import urllib.parse
+The implementation now lives in :mod:`a2a.utils.url_validation` as a
+composition of :class:`a2a.utils.url_validation.UrlValidator` rules
+(issue #1023); this module keeps the historical boolean, fail-closed
+entry point.
+"""
 
+from a2a.utils.url_validation import PushNotificationUrlValidator
 
-logger = logging.getLogger(__name__)
 
-
-def _ip_is_blocked(ip_str: str) -> bool:
-    """Whether an address is not a public unicast destination."""
-    try:
-        addr = ipaddress.ip_address(ip_str.split('%', maxsplit=1)[0])
-    except ValueError:
-        return True
-    return (
-        addr.is_private
-        or addr.is_loopback
-        or addr.is_link_local
-        or addr.is_multicast
-        or addr.is_reserved
-        or addr.is_unspecified
-    )
+_validator = PushNotificationUrlValidator()
 
 
 async def validate_push_notification_url(url: str) -> bool:
@@ -46,38 +32,4 @@ async def validate_push_notification_url(url: str) -> bool:
     The default on those constructors is ``None`` (no library
     screening).
     """
-    try:
-        parsed = urllib.parse.urlparse(url)
-        explicit_port = parsed.port
-    except ValueError:
-        logger.warning('Push-notification URL is unparseable: %s', url)
-        return False
-    if parsed.scheme not in ('http', 'https'):
-        logger.warning(
-            'Push-notification URL scheme %r is not http/https: %s',
-            parsed.scheme,
-            url,
-        )
-        return False
-    host = parsed.hostname
-    if not host:
-        logger.warning('Push-notification URL has no hostname: %s', url)
-        return False
-    port = explicit_port or (443 if parsed.scheme == 'https' else 80)
-    try:
-        loop = asyncio.get_running_loop()
-        infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
-    except OSError:
-        logger.warning(
-            'Push-notification host %r could not be resolved: %s', host, url
-        )
-        return False
-    for info in infos:
-        if _ip_is_blocked(str(info[4][0])):
-            logger.warning(
-                'Push-notification host %r resolves to a non-public address: %s',
-                host,
-                url,
-            )
-            return False
-    return True
+    return await _validator.validate(url)
diff --git a/src/a2a/utils/url_validation.py b/src/a2a/utils/url_validation.py
new file mode 100644
index 000000000..a7fa05b22
--- /dev/null
+++ b/src/a2a/utils/url_validation.py
@@ -0,0 +1,243 @@
+"""Composable, domain-agnostic URL validation.
+
+Foundation for screening URLs that the SDK fetches or dials (agent card
+endpoints, push-notification webhooks) before invocation, so that
+user-controlled URLs cannot reach loopback, private, or otherwise
+non-public destinations (SSRF).
+
+The design follows the sketch agreed on issue #1023: a core
+:class:`UrlValidator` parses and (optionally) resolves the URL, runs a
+sequence of composable :class:`UrlValidationRule` objects, and returns
+the resolved URL — including the addresses the host resolved to, so
+callers can pin the connection to a validated address. Domain-specific
+wrappers (e.g.
+:func:`a2a.utils.push_url_validator.validate_push_notification_url`)
+own which fields to validate and how failures surface.
+"""
+
+import asyncio
+import ipaddress
+import logging
+import socket
+import urllib.parse
+
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable, Callable, Sequence
+from dataclasses import dataclass
+
+
+logger = logging.getLogger(__name__)
+
+UrlAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
+
+Resolver = Callable[[str, int], Awaitable[list[tuple]]]
+
+
+class InvalidUrlError(ValueError):
+    """Raised by UrlValidator.validate when a URL is rejected."""
+
+
+@dataclass(frozen=True)
+class ResolvedUrl:
+    """A parsed URL together with the addresses its host resolved to."""
+
+    raw: str
+    parsed: urllib.parse.SplitResult
+    addresses: tuple[UrlAddress, ...]
+
+    @property
+    def hostname(self) -> str:
+        """The lowercased host from the parsed URL (may be empty)."""
+        return self.parsed.hostname or ''
+
+
+def _is_non_public(address: UrlAddress) -> bool:
+    """Whether an address is not a public unicast destination."""
+    return (
+        address.is_private
+        or address.is_loopback
+        or address.is_link_local
+        or address.is_multicast
+        or address.is_reserved
+        or address.is_unspecified
+    )
+
+
+class UrlValidationRule(ABC):
+    """A single composable URL check.
+
+    Implementations raise :class:`InvalidUrlError` to reject the URL;
+    returning normally means 'continue to the next rule'.
+    """
+
+    @abstractmethod
+    async def check(self, url: ResolvedUrl) -> None:
+        """Check the resolved URL. Raise InvalidUrlError to reject."""
+
+
+class RequireScheme(UrlValidationRule):
+    """Rejects URLs whose scheme is not one of the allowed values."""
+
+    def __init__(self, schemes: Sequence[str]) -> None:
+        self._schemes = tuple(scheme.lower() for scheme in schemes)
+
+    async def check(self, url: ResolvedUrl) -> None:
+        """Reject the URL when its scheme is not allowed."""
+        if url.parsed.scheme not in self._schemes:
+            raise InvalidUrlError(
+                f'Scheme {url.parsed.scheme!r} is not allowed; expected one '
+                f'of: {", ".join(self._schemes)}'
+            )
+
+
+class BlockPrivateNetworks(UrlValidationRule):
+    """Rejects hosts resolving to non-public addresses.
+
+    Blocks loopback, private, link-local, multicast, reserved, and
+    unspecified destinations (e.g. 169.254.169.254 cloud metadata,
+    internal services). Deployments that legitimately use private
+    networks can exempt specific hostnames via ``allow_hosts`` or whole
+    ranges via ``allow_cidrs``.
+    """
+
+    def __init__(
+        self,
+        *,
+        allow_hosts: Sequence[str] = (),
+        allow_cidrs: Sequence[str] = (),
+    ) -> None:
+        # Hostnames are matched case-insensitively: urlparse lowercases
+        # parsed hostnames, so the exemptions are normalized too.
+        self._allow_hosts = frozenset(host.lower() for host in allow_hosts)
+        self._allow_cidrs = tuple(
+            ipaddress.ip_network(cidr, strict=False) for cidr in allow_cidrs
+        )
+
+    def _is_allowed(self, url: ResolvedUrl, address: UrlAddress) -> bool:
+        if url.hostname in self._allow_hosts:
+            return True
+        return any(address in network for network in self._allow_cidrs)
+
+    async def check(self, url: ResolvedUrl) -> None:
+        """Reject the URL when any resolved address is non-public."""
+        for address in url.addresses:
+            if self._is_allowed(url, address):
+                continue
+            if _is_non_public(address):
+                raise InvalidUrlError(
+                    f'Host {url.hostname!r} resolves to a non-public '
+                    f'address: {address}'
+                )
+
+
+async def _default_resolver(host: str, port: int) -> list[tuple]:
+    loop = asyncio.get_running_loop()
+    return await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+
+
+class UrlValidator:
+    """Parses and optionally resolves a URL, then runs composable rules.
+
+    Rules run in order; the first rule that raises :class:`InvalidUrlError`
+    rejects the URL. With ``resolve=True`` (default) the host is resolved
+    via the running event loop's ``getaddrinfo`` before rules run, so
+    IP-based rules (e.g. :class:`BlockPrivateNetworks`) see every address
+    the host maps to — and resolution failures fail closed.
+    """
+
+    def __init__(
+        self,
+        rules: Sequence[UrlValidationRule] = (),
+        *,
+        resolve: bool = True,
+        resolver: Resolver | None = None,
+    ) -> None:
+        self._rules = tuple(rules)
+        self._resolve = resolve
+        self._resolver = resolver or _default_resolver
+
+    async def validate(self, url: str) -> ResolvedUrl:
+        """Validate ``url`` and return it with its resolved addresses."""
+        try:
+            parsed = urllib.parse.urlsplit(url)
+            port = parsed.port or (443 if parsed.scheme == 'https' else 80)
+        except ValueError as e:
+            raise InvalidUrlError(f'URL is unparseable: {url}') from e
+        host = parsed.hostname
+        if not host:
+            raise InvalidUrlError(f'URL has no hostname: {url}')
+
+        addresses: tuple[UrlAddress, ...] = ()
+        if self._resolve:
+            try:
+                infos = await self._resolver(host, port)
+            except OSError as e:
+                raise InvalidUrlError(
+                    f'Host {host!r} could not be resolved: {url}'
+                ) from e
+            resolved: list[UrlAddress] = []
+            for info in infos:
+                raw_ip = str(info[4][0]).split('%', maxsplit=1)[0]
+                try:
+                    resolved.append(ipaddress.ip_address(raw_ip))
+                except ValueError as e:
+                    raise InvalidUrlError(
+                        f'Host {host!r} resolved to an unparseable address'
+                    ) from e
+            # getaddrinfo can return the same address several times (CNAME
+            # chains, multi-record families); deduplicate while keeping the
+            # resolver's order.
+            addresses = tuple(dict.fromkeys(resolved))
+
+        resolved_url = ResolvedUrl(raw=url, parsed=parsed, addresses=addresses)
+        for rule in self._rules:
+            await rule.check(resolved_url)
+        return resolved_url
+
+
+class PushNotificationUrlValidator:
+    """Domain wrapper validating push-notification webhook URLs.
+
+    Keeps the boolean, fail-closed contract of
+    :func:`a2a.utils.push_url_validator.validate_push_notification_url`.
+    """
+
+    def __init__(self) -> None:
+        self._validator = UrlValidator(
+            rules=(RequireScheme(('http', 'https')), BlockPrivateNetworks())
+        )
+
+    async def validate(self, url: str) -> bool:
+        """Return True if the webhook URL is safe to fetch."""
+        try:
+            await self._validator.validate(url)
+        except InvalidUrlError as e:
+            logger.warning('Push-notification URL rejected: %s (%s)', url, e)
+            return False
+        return True
+
+
+class AgentCardUrlValidator:
+    """Domain wrapper validating URLs advertised on an agent card.
+
+    Covers the service endpoint and additional interfaces, screening
+    them before the SDK dials them. By default only http/https on
+    public addresses are allowed; deployments that legitimately target
+    private networks can pass ``allow_private_networks=True``.
+    """
+
+    def __init__(
+        self,
+        *,
+        allowed_schemes: Sequence[str] = ('http', 'https'),
+        allow_private_networks: bool = False,
+        resolver: Resolver | None = None,
+    ) -> None:
+        rules: list[UrlValidationRule] = [RequireScheme(allowed_schemes)]
+        if not allow_private_networks:
+            rules.append(BlockPrivateNetworks())
+        self._validator = UrlValidator(rules, resolver=resolver)
+
+    async def validate(self, url: str) -> ResolvedUrl:
+        """Validate and return the resolved URL (addresses included)."""
+        return await self._validator.validate(url)
diff --git a/tests/utils/test_push_url_validator.py b/tests/utils/test_push_url_validator.py
new file mode 100644
index 000000000..263c6e9f2
--- /dev/null
+++ b/tests/utils/test_push_url_validator.py
@@ -0,0 +1,57 @@
+"""Tests for the validate_push_notification_url compatibility wrapper.
+
+The wrapper delegates to PushNotificationUrlValidator and keeps the
+historical boolean, fail-closed contract (issue #1023 refactor).
+"""
+
+import a2a.utils.push_url_validator as push_url_validator_module
+import pytest
+
+from a2a.utils.push_url_validator import validate_push_notification_url
+
+
+PUBLIC_IP = '93.184.216.34'
+
+
+def _pin_resolver(monkeypatch, *ips: str) -> None:
+    """Pin the module singleton's resolver to deterministic addresses."""
+
+    async def resolve(host: str, port: int):
+        return [(2, 1, 6, '', (ip, port)) for ip in ips]
+
+    monkeypatch.setattr(
+        push_url_validator_module._validator._validator,
+        '_resolver',
+        resolve,
+    )
+
+
+def _pin_failing_resolver(monkeypatch) -> None:
+    """Pin the module singleton's resolver to a failing lookup."""
+
+    async def resolve(host: str, port: int):
+        raise OSError('no DNS')
+
+    monkeypatch.setattr(
+        push_url_validator_module._validator._validator,
+        '_resolver',
+        resolve,
+    )
+
+
+@pytest.mark.asyncio
+async def test_returns_true_for_public_http_url(monkeypatch):
+    _pin_resolver(monkeypatch, PUBLIC_IP)
+    assert await validate_push_notification_url('http://example.com/hook')
+
+
+@pytest.mark.asyncio
+async def test_returns_false_for_non_http_scheme(monkeypatch):
+    _pin_resolver(monkeypatch, PUBLIC_IP)
+    assert not await validate_push_notification_url('ftp://example.com/file')
+
+
+@pytest.mark.asyncio
+async def test_returns_false_when_resolution_fails(monkeypatch):
+    _pin_failing_resolver(monkeypatch)
+    assert not await validate_push_notification_url('http://example.com/hook')
diff --git a/tests/utils/test_url_validation.py b/tests/utils/test_url_validation.py
new file mode 100644
index 000000000..d43e11c29
--- /dev/null
+++ b/tests/utils/test_url_validation.py
@@ -0,0 +1,333 @@
+"""Tests for the composable URL validation primitives (issue #1023)."""
+
+import asyncio
+
+import pytest
+
+from a2a.utils.url_validation import (
+    AgentCardUrlValidator,
+    BlockPrivateNetworks,
+    InvalidUrlError,
+    RequireScheme,
+    UrlValidator,
+)
+
+
+PUBLIC_IP = '93.184.216.34'
+
+
+def _resolver_with(*ips: str):
+    async def resolve(host: str, port: int):
+        return [(2, 1, 6, '', (ip, port)) for ip in ips]
+
+    return resolve
+
+
+def _resolver_failing():
+    async def resolve(host: str, port: int):
+        raise OSError('no DNS')
+
+    return resolve
+
+
+async def _validate(
+    url: str,
+    *,
+    rules=None,
+    ips: tuple[str, ...] = (PUBLIC_IP,),
+    resolve: bool = True,
+    resolver=None,
+):
+    if rules is None:
+        rules = (RequireScheme(('http', 'https')), BlockPrivateNetworks())
+    if resolver is None:
+        resolver = _resolver_with(*ips)
+    validator = UrlValidator(rules, resolve=resolve, resolver=resolver)
+    return await validator.validate(url)
+
+
+@pytest.mark.asyncio
+async def test_public_http_url_passes_and_returns_resolved_url():
+    resolved = await _validate('http://example.com/callback')
+
+    assert resolved.raw == 'http://example.com/callback'
+    assert resolved.parsed.hostname == 'example.com'
+    assert [str(a) for a in resolved.addresses] == [PUBLIC_IP]
+
+
+@pytest.mark.asyncio
+async def test_resolver_receives_default_port_by_scheme():
+    seen_ports = []
+
+    async def recording_resolver(host: str, port: int):
+        seen_ports.append(port)
+        return [(2, 1, 6, '', (PUBLIC_IP, port))]
+
+    validator = UrlValidator(
+        (RequireScheme(('http', 'https')),), resolver=recording_resolver
+    )
+    await validator.validate('https://example.com/hook')
+    await validator.validate('http://example.com/hook')
+
+    assert seen_ports == [443, 80]
+
+
+@pytest.mark.asyncio
+async def test_resolver_receives_explicit_port():
+    seen_ports = []
+
+    async def recording_resolver(host: str, port: int):
+        seen_ports.append(port)
+        return [(2, 1, 6, '', (PUBLIC_IP, port))]
+
+    validator = UrlValidator(
+        (RequireScheme(('http', 'https')),), resolver=recording_resolver
+    )
+    await validator.validate('http://example.com:8080/hook')
+
+    assert seen_ports == [8080]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'ip',
+    [
+        '127.0.0.1',
+        '10.0.0.5',
+        '172.16.1.9',
+        '192.168.0.20',
+        '169.254.169.254',
+        '224.0.0.1',
+        '240.0.0.1',
+        '0.0.0.0',
+    ],
+)
+async def test_block_private_networks_rejects_non_public_ips(ip: str):
+    with pytest.raises(InvalidUrlError, match='non-public'):
+        await _validate(f'http://example.com/hook', ips=(ip,))
+
+
+@pytest.mark.asyncio
+async def test_block_private_networks_allow_hosts_exception():
+    rules = (
+        RequireScheme(('http', 'https')),
+        BlockPrivateNetworks(allow_hosts=('internal.corp',)),
+    )
+    resolved = await _validate(
+        'http://internal.corp/api', rules=rules, ips=('10.1.2.3',)
+    )
+
+    assert resolved.hostname == 'internal.corp'
+
+
+@pytest.mark.asyncio
+async def test_block_private_networks_allow_cidrs_exception():
+    rules = (
+        RequireScheme(('http', 'https')),
+        BlockPrivateNetworks(allow_cidrs=('10.1.2.0/24',)),
+    )
+    resolved = await _validate(
+        'http://internal.corp/api', rules=rules, ips=('10.1.2.3',)
+    )
+
+    assert resolved.hostname == 'internal.corp'
+
+
+@pytest.mark.asyncio
+async def test_block_private_networks_allow_cidrs_still_blocks_outside():
+    rules = (
+        RequireScheme(('http', 'https')),
+        BlockPrivateNetworks(allow_cidrs=('10.1.2.0/24',)),
+    )
+    with pytest.raises(InvalidUrlError, match='non-public'):
+        await _validate(
+            'http://internal.corp/api', rules=rules, ips=('10.9.9.9',)
+        )
+
+
+@pytest.mark.asyncio
+async def test_block_private_networks_allow_hosts_is_case_insensitive():
+    """Allowlist matching must survive host casing: parsed hostnames are
+    lowercased by urlparse, so configured exemptions are normalized too."""
+    rules = (
+        RequireScheme(('http', 'https')),
+        BlockPrivateNetworks(allow_hosts=('Internal.Corp',)),
+    )
+    resolved = await _validate(
+        'http://INTERNAL.corp/api', rules=rules, ips=('10.1.2.3',)
+    )
+
+    assert resolved.hostname == 'internal.corp'
+
+
+@pytest.mark.asyncio
+async def test_duplicate_resolved_addresses_are_deduplicated():
+    resolved = await _validate(
+        'http://example.com/hook', ips=(PUBLIC_IP, PUBLIC_IP)
+    )
+
+    assert [str(a) for a in resolved.addresses] == [PUBLIC_IP]
+
+
+@pytest.mark.asyncio
+async def test_block_private_networks_ignores_public_addresses():
+    resolved = await _validate('http://example.com/hook')
+
+    assert resolved.hostname == 'example.com'
+
+
+@pytest.mark.asyncio
+async def test_require_scheme_rejects_other_schemes():
+    with pytest.raises(InvalidUrlError, match="[Ss]cheme 'ftp'"):
+        await _validate('ftp://example.com/file')
+
+
+@pytest.mark.asyncio
+async def test_require_scheme_is_case_insensitive():
+    resolved = await _validate('HTTP://example.com/hook')
+
+    assert resolved.parsed.scheme == 'http'
+
+
+@pytest.mark.asyncio
+async def test_missing_scheme_is_rejected():
+    with pytest.raises(InvalidUrlError):
+        await _validate('example.com/hook')
+
+
+@pytest.mark.asyncio
+async def test_unresolvable_host_fails_closed():
+    with pytest.raises(InvalidUrlError, match='could not be resolved'):
+        await _validate(
+            'http://does-not-resolve.invalid/',
+            resolver=_resolver_failing(),
+        )
+
+
+@pytest.mark.asyncio
+async def test_unparseable_url_is_rejected():
+    with pytest.raises(InvalidUrlError, match='unparseable'):
+        await _validate('http://[invalid-ipv6/hook')
+
+
+@pytest.mark.asyncio
+async def test_invalid_port_is_rejected():
+    with pytest.raises(InvalidUrlError, match='unparseable'):
+        await _validate('http://example.com:99999/hook')
+
+
+@pytest.mark.asyncio
+async def test_missing_hostname_is_rejected():
+    with pytest.raises(InvalidUrlError, match='no hostname'):
+        await _validate('http:///path')
+
+
+@pytest.mark.asyncio
+async def test_resolve_false_skips_ip_rules_but_keeps_scheme_rules():
+    rules = (RequireScheme(('http', 'https')), BlockPrivateNetworks())
+    resolved = await _validate(
+        'http://internal.corp/api',
+        rules=rules,
+        ips=('10.9.9.9',),
+        resolve=False,
+    )
+
+    assert resolved.addresses == ()
+    assert resolved.hostname == 'internal.corp'
+
+
+@pytest.mark.asyncio
+async def test_rules_run_in_order_and_first_rejection_wins():
+    order = []
+
+    class RecordingRule(RequireScheme):
+        async def check(self, url):
+            order.append(self.__class__.__name__)
+            await super().check(url)
+
+    class SecondRule(BlockPrivateNetworks):
+        async def check(self, url):
+            order.append('SecondRule')
+            await super().check(url)
+
+    validator = UrlValidator(
+        (
+            RecordingRule(('http', 'https')),
+            SecondRule(),
+        ),
+        resolver=_resolver_with('127.0.0.1'),
+    )
+    with pytest.raises(InvalidUrlError):
+        await validator.validate('http://example.com/hook')
+
+    assert order[0] == 'RecordingRule'
+
+
+@pytest.mark.asyncio
+async def test_agent_card_url_validator_accepts_public_endpoint():
+    validator = AgentCardUrlValidator(resolver=_resolver_with(PUBLIC_IP))
+    resolved = await validator.validate('https://agent.example.com/card')
+
+    assert resolved.hostname == 'agent.example.com'
+    assert [str(a) for a in resolved.addresses] == [PUBLIC_IP]
+
+
+@pytest.mark.asyncio
+async def test_agent_card_url_validator_blocks_private_by_default():
+    validator = AgentCardUrlValidator(resolver=_resolver_with('10.0.0.9'))
+    with pytest.raises(InvalidUrlError, match='non-public'):
+        await validator.validate('http://internal.agent:8080/card')
+
+
+@pytest.mark.asyncio
+async def test_agent_card_url_validator_allows_private_when_opted_in():
+    validator = AgentCardUrlValidator(
+        allow_private_networks=True, resolver=_resolver_with('10.0.0.9')
+    )
+    resolved = await validator.validate('http://internal.agent:8080/card')
+
+    assert resolved.hostname == 'internal.agent'
+
+
+@pytest.mark.asyncio
+async def test_agent_card_url_validator_supports_https_only():
+    validator = AgentCardUrlValidator(
+        allowed_schemes=('https',), resolver=_resolver_with(PUBLIC_IP)
+    )
+    with pytest.raises(InvalidUrlError, match="[Ss]cheme 'http'"):
+        await validator.validate('http://agent.example.com/card')
+    resolved = await validator.validate('https://agent.example.com/card')
+
+    assert resolved.hostname == 'agent.example.com'
+
+
+@pytest.mark.asyncio
+async def test_ipv4_mapped_ipv6_loopback_is_rejected():
+    with pytest.raises(InvalidUrlError, match='non-public'):
+        await _validate('http://example.com/hook', ips=('::ffff:127.0.0.1',))
+
+
+@pytest.mark.asyncio
+async def test_multiple_resolved_addresses_any_non_public_rejects():
+    with pytest.raises(InvalidUrlError, match='non-public'):
+        await _validate(
+            'http://dual-homed.example/', ips=(PUBLIC_IP, '192.168.1.1')
+        )
+
+
+@pytest.mark.asyncio
+async def test_validate_can_be_cancelled_while_resolving():
+    validator = UrlValidator(
+        (RequireScheme(('http', 'https')),),
+        resolver=_resolver_with(PUBLIC_IP),
+    )
+
+    async def never_returning_resolver(host: str, port: int):
+        await asyncio.Event().wait()
+
+    validator._resolver = never_returning_resolver  # type: ignore[assignment]
+    task = asyncio.create_task(validator.validate('http://example.com/'))
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task