Connect to Restate Cloud without a port: restate.connect() and python -m restate connect

Cargo.toml

@@ -13,5 +13,22 @@ doc = false
 
 [dependencies]
 pyo3 = { version = "0.30.1", features = ["extension-module"] }
+pyo3-async-runtimes = { version = "0.30", features = ["tokio-runtime"] }
+tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
-restate-sdk-shared-core = { version = "8.2.0", features = ["request_identity", "sha2_random_seed", "rust_crypto"] }
+restate-sdk-shared-core = { version = "8.2.0", features = ["request_identity", "sha2_random_seed", "rust_crypto", "tunnel"] }
+
+# ASGI bridge for the tunnel
+bytes = "1.10"
+http = "1.3"
+http-body = "1.0"
+http-body-util = "0.1"
+hyper = { version = "1.8", default-features = false }
+percent-encoding = "2.3"
+tokio = { version = "1.49", default-features = false, features = ["sync", "rt-multi-thread"] }
+tokio-stream = { version = "0.1", default-features = false }
+tokio-util = { version = "0.7", default-features = false }
+
+# Until restate-sdk-shared-core with the `cloud` and `tunnel` features is released.
+[patch.crates-io]
+restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core", branch = "cloud-exp-revamp" }

README.md

@@ -26,6 +26,34 @@ To use this SDK, add the dependency to your project:
 pip install restate_sdk
 ```
 
+## Connecting to Restate Cloud
+
+An application can connect outbound to Restate Cloud instead of exposing a port. Copy a connect key
+from the Cloud UI (Development → Configure SDK), pick a name for the deployment, and run:
+
+```shell
+RESTATE_CONNECT_KEY=rstcloudv1_... python -m restate connect app:app --name greeter
+```
+
+where `app:app` is the module and attribute holding the application returned by `restate.app(...)`.
+Then register the deployment as `connect://greeter`. The same from code, running until stopped:
+
+```python
+if __name__ == "__main__":
+    asyncio.run(restate.connect(app, name="greeter"))
+```
+
+To do something while connected, use the context manager:
+
+```python
+async with restate.with_connection(app, name="greeter") as connection:
+    print(connection.info.deployment_url)
+    await connection.wait()
+```
+
+Clients can read an ingress key the same way: `restate.create_client()` without an URL uses
+`RESTATE_INGRESS_KEY` (Development → Configure Client).
+
 ## Versions
 
 The compatibility with Restate is described in the following table. sdk-python versions < 0.10 are deprecated and no longer supported.

python/restate/__init__.py

@@ -43,6 +43,7 @@
 from .asyncio import as_completed, gather, wait_completed, select
 
 from .endpoint import app
+from .connection import connect, with_connection, ConnectError
 
 from .logging import getLogger, RestateLoggingFilter
 
@@ -83,7 +84,7 @@ def test_harness(
 
     @asynccontextmanager
     async def create_client(
-        ingress: str,
+        ingress: typing.Optional[str] = None,
         headers: typing.Optional[dict] = None,
         journal_value_codec: typing.Optional[JournalValueCodec] = None,
     ) -> typing.AsyncGenerator[RestateClient, None]:
@@ -112,6 +113,9 @@ async def create_client(
     "JournalValueCodec",
     "JournalValueCodecProvider",
     "app",
+    "connect",
+    "with_connection",
+    "ConnectError",
     "create_test_harness",
     "test_harness",
     "HarnessEnvironment",

python/restate/__main__.py

@@ -0,0 +1,93 @@
+#
+#  Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH
+#
+#  This file is part of the Restate SDK for Python,
+#  which is released under the MIT license.
+#
+#  You can find a copy of the license in file LICENSE in the root
+#  directory of this repository or package, or at
+#  https://github.com/restatedev/sdk-typescript/blob/main/LICENSE
+#
+"""
+``python -m restate``: command-line entry points.
+
+    python -m restate connect app:app --name greeter
+
+serves the ASGI application ``app`` of module ``app`` through an outbound
+connection to Restate Cloud, until SIGINT or SIGTERM.
+"""
+
+import argparse
+import asyncio
+import importlib
+import os
+import sys
+import typing
+
+from restate.connection import ConnectError, connect
+
+EXIT_OK = 0
+EXIT_FAILURE = 1
+EXIT_USAGE = 2
+
+
+def _load_app(spec: str) -> typing.Any:
+    """Import ``module[:attribute]`` (attribute defaults to ``app``), like hypercorn."""
+    module_name, _, attribute = spec.partition(":")
+    if not module_name:
+        raise ValueError(f"invalid application {spec!r}: expected module:attribute")
+    attribute = attribute or "app"
+    try:
+        module = importlib.import_module(module_name)
+    except ImportError as error:
+        raise ValueError(f"cannot import module {module_name!r}: {error}") from error
+    try:
+        return getattr(module, attribute)
+    except AttributeError as error:
+        raise ValueError(f"module {module_name!r} has no attribute {attribute!r}") from error
+
+
+def main(argv: typing.Optional[typing.Sequence[str]] = None) -> int:
+    """Entry point of ``python -m restate``."""
+    parser = argparse.ArgumentParser(prog="python -m restate", description="Restate SDK command line.")
+    commands = parser.add_subparsers(dest="command", required=True)
+    connect_parser = commands.add_parser(
+        "connect",
+        help="serve an application through an outbound connection to Restate Cloud",
+        description=(
+            "Serve the ASGI application through an outbound connection to Restate Cloud, so it needs "
+            "no inbound port. The connect key comes from the Cloud UI (Development → Configure SDK)."
+        ),
+    )
+    connect_parser.add_argument(
+        "app",
+        help="the application as module:attribute, for example app:app (attribute defaults to app)",
+    )
+    connect_parser.add_argument(
+        "--name",
+        help="the deployment name, registered as connect://<name> (default: $RESTATE_CONNECT_NAME)",
+    )
+    connect_parser.add_argument(
+        "--key",
+        help="the connect key (default: $RESTATE_CONNECT_KEY; prefer the variable over the flag)",
+    )
+    args = parser.parse_args(argv)
+
+    # Resolve the application from the working directory, like a server would.
+    sys.path.insert(0, os.getcwd())
+    try:
+        app = _load_app(args.app)
+    except ValueError as error:
+        parser.error(str(error))
+        return EXIT_USAGE  # pragma: no cover - parser.error exits
+
+    try:
+        asyncio.run(connect(app, name=args.name, key=args.key))
+    except ConnectError as error:
+        print(f"error: {error}", file=sys.stderr, flush=True)
+        return EXIT_FAILURE
+    return EXIT_OK
+
+
+if __name__ == "__main__":
+    sys.exit(main())

python/restate/client.py

@@ -481,21 +481,48 @@ async def workflow_send(
         return RestateClientSendHandle(send.get("invocationId", ""), 200)
 
 
+def resolve_ingress(
+    ingress: typing.Optional[str], headers: typing.Optional[dict]
+) -> typing.Tuple[str, typing.Optional[dict]]:
+    """
+    The ingress URL and default headers: the given ones, or the URL and bearer token of
+    ``RESTATE_INGRESS_KEY`` (an ingress key from the Cloud UI, Development → Configure Client).
+    """
+    if ingress:
+        return ingress, headers
+    # pylint: disable=import-error,no-name-in-module,import-outside-toplevel
+    from restate._internal import INGRESS_KEY_URL, ConnectError, IngressKey
+
+    key = IngressKey.from_env()
+    if key is None:
+        raise ConnectError(
+            "create_client(): ingress is required — pass the ingress URL, or set RESTATE_INGRESS_KEY "
+            f"with an ingress key from {INGRESS_KEY_URL}"
+        )
+    merged = dict(headers or {})
+    if not any(name.lower() == "authorization" for name in merged):
+        merged["Authorization"] = key.bearer()
+    return key.ingress_url(), merged
+
+
 @asynccontextmanager
 async def create_client(
-    ingress: str,
+    ingress: typing.Optional[str] = None,
     headers: typing.Optional[dict] = None,
     journal_value_codec: typing.Optional[JournalValueCodec] = None,
 ) -> typing.AsyncGenerator[RestateClient, None]:
     """
     Create a new Restate client.
 
     Args:
-        ingress: The base URL of the Restate ingress.
+        ingress: The base URL of the Restate ingress. When omitted, the URL and the bearer
+            token come from ``RESTATE_INGRESS_KEY``, an ingress key from the Cloud UI
+            (Development → Configure Client).
         headers: Optional default headers to send with every request.
         journal_value_codec: Optional journal value codec. When set, request bodies are encoded and
             call responses decoded through it. It must match the codec configured on the endpoint.
             NOTE: This is experimental and may change in future releases.
     """
+    ingress, headers = resolve_ingress(ingress, headers)
     async with httpx.AsyncClient(base_url=ingress, headers=headers, http2=True) as http_client:
         yield Client(http_client, headers, journal_codec=journal_value_codec)