@@ -0,0 +1 @@
+Expose `connect_timeout` on `Client.connect` and `ConnectConfig` to bound TCP connection attempts independently of RPC timeouts. The default remains unset.
@@ -85,6 +85,7 @@ class ClientConfig:
     grpc_compression: str
     payloads_warn_size: int
     memo_warn_size: int
+    connect_timeout_millis: int | None = None
 
 
 @dataclass
 
@@ -44,6 +44,7 @@ pub struct ClientConfig {
     tls_config: Option<ClientTlsConfig>,
     retry_config: Option<ClientRetryConfig>,
     keep_alive_config: Option<ClientKeepAliveConfig>,
+    connect_timeout_millis: Option<u64>,
     http_connect_proxy_config: Option<ClientHttpConnectProxyConfig>,
     dns_load_balancing_config: Option<ClientDnsLoadBalancingConfig>,
     grpc_compression: String,
@@ -277,6 +278,7 @@ impl ClientConfig {
                 .map_or(RetryOptions::default(), |c| c.into()),
         )
         .keep_alive(self.keep_alive_config.map(Into::into))
+        .maybe_connect_timeout(self.connect_timeout_millis.map(Duration::from_millis))
         .maybe_http_connect_proxy(self.http_connect_proxy_config.map(Into::into))
         .dns_load_balancing(dns_load_balancing)
         .grpc_compression(grpc_compression_from_str(&self.grpc_compression)?)
 
@@ -146,6 +146,7 @@ async def connect(
         tls: bool | TLSConfig | None = None,
         retry_config: RetryConfig | None = None,
         keep_alive_config: KeepAliveConfig | None = KeepAliveConfig.default,
+        connect_timeout: timedelta | None = None,
         rpc_metadata: Mapping[str, str | bytes] = {},
         identity: str | None = None,
         lazy: bool = False,
@@ -200,6 +201,10 @@ async def connect(
                 connection. Default is to check every 30s and kill the
                 connection if a response doesn't come back in 15s. Can be set to
                 ``None`` to disable.
+            connect_timeout: Maximum duration of a TCP connection attempt.
+                If unset, no explicit connection timeout is applied. This is
+                separate from RPC timeouts and does not bound the total duration
+                of a call with retries.
             rpc_metadata: Headers to use for all calls to the server. Keys here
                 can be overriden by per-call RPC metadata keys.
             identity: Identity for this client. If unset, a default is created
@@ -227,6 +232,7 @@ async def connect(
             tls=tls,
             retry_config=retry_config,
             keep_alive_config=keep_alive_config,
+            connect_timeout=connect_timeout,
             rpc_metadata=rpc_metadata,
             identity=identity or "",
             lazy=lazy,
@@ -3013,6 +3019,7 @@ class ClientConnectConfig(TypedDict, total=False):
     tls: bool | TLSConfig | None
     retry_config: RetryConfig | None
     keep_alive_config: KeepAliveConfig | None
+    connect_timeout: timedelta | None
     rpc_metadata: Mapping[str, str | bytes]
     identity: str | None
     lazy: bool
 
@@ -239,6 +239,7 @@ class ConnectConfig:
     dns_load_balancing_config: DnsLoadBalancingConfig | None = None
     grpc_compression: GrpcCompression = GrpcCompression.GZIP
     payload_limits: PayloadLimitsConfig = field(default_factory=PayloadLimitsConfig)
+    connect_timeout: timedelta | None = None
 
     def __post_init__(self) -> None:
         """Set extra defaults on unset properties."""
@@ -287,6 +288,11 @@ def _to_bridge_config(self) -> temporalio.bridge.client.ClientConfig:
                 if self.keep_alive_config
                 else None
             ),
+            connect_timeout_millis=(
+                int(self.connect_timeout.total_seconds() * 1000)
+                if self.connect_timeout is not None
+                else None
+            ),
             metadata=self.rpc_metadata,
             identity=self.identity,
             client_name="temporal-python",
 
@@ -1,6 +1,8 @@
+import asyncio
 import inspect
 import os
 import re
+import socket
 from collections.abc import Callable, Mapping
 from datetime import timedelta
 from typing import Any
@@ -331,3 +333,58 @@ async def test_service(
         proto_service="Health",
         target_service_name="health_service",
     )
+
+
+@pytest.mark.parametrize("timeout", [None, timedelta(0), timedelta(milliseconds=1250)])
+def test_connect_config_connect_timeout(timeout: timedelta | None):
+    config = temporalio.service.ConnectConfig(
+        target_host="localhost:7233", connect_timeout=timeout
+    )
+    assert config._to_bridge_config().connect_timeout_millis == (
+        None if timeout is None else int(timeout.total_seconds() * 1000)
+    )
+
+
+async def test_connect_timeout_bounds_tcp_connection():
+    listener = socket.socket()
+    connections: list[socket.socket] = []
+    try:
+        listener.bind(("127.0.0.1", 0))
+        listener.listen(1)
+        address = listener.getsockname()
+        loop = asyncio.get_running_loop()
+        # Fill the accept queue without accepting connections, so subsequent TCP
+        # attempts stall locally instead of depending on an unreachable remote IP.
+        for _ in range(128):
+            connection = socket.socket()
+            connections.append(connection)
+            connection.setblocking(False)
+            try:
+                await asyncio.wait_for(loop.sock_connect(connection, address), 0.1)
+            except asyncio.TimeoutError:
+                break
+        else:
+            pytest.fail("Could not fill the listener's accept queue")
+
+        with pytest.raises(RuntimeError, match="tcp connect error.*TimedOut"):
+            await asyncio.wait_for(
+                Client.connect(
+                    f"127.0.0.1:{address[1]}",
+                    connect_timeout=timedelta(milliseconds=100),
+                ),
+                timeout=5,
+            )
+    finally:
+        for connection in connections:
+            connection.close()
+        listener.close()
+
+
+async def test_connect_timeout_allows_healthy_connection(client: Client):
+    connected = await Client.connect(
+        client.service_client.config.target_host,
+        connect_timeout=timedelta(seconds=5),
+    )
+    await connected.workflow_service.get_system_info(
+        temporalio.api.workflowservice.v1.GetSystemInfoRequest()
+    )
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	+Expose `connect_timeout` on `Client.connect` and `ConnectConfig` to bound TCP connection attempts independently of RPC timeouts. The default remains unset.