Enable no-transport mode as startup parameter

aries_cloudagent/config/argparse.py

@@ -683,8 +683,10 @@ def get_settings(self, args: Namespace) -> dict:
         if args.endpoint:
             settings["default_endpoint"] = args.endpoint[0]
             settings["additional_endpoints"] = args.endpoint[1:]
-        else:
+
+        elif "no_transport" not in args:
             raise ArgsParseError("-e/--endpoint is required")
+
         if args.profile_endpoint:
             settings["profile_endpoint"] = args.profile_endpoint
 
@@ -1273,6 +1275,19 @@ class TransportGroup(ArgumentGroup):
 
     def add_arguments(self, parser: ArgumentParser):
         """Add transport-specific command line arguments to the parser."""
+        parser.add_argument(
+            "--no-transport",
+            dest="no_transport",
+            action="store_true",
+            env_var="ACAPY_NO_TRANSPORT",
+            help=(
+                "Specifies that aca-py will run with no transport configured. "
+                "This must be set if running in no-transport mode.  Overrides any "
+                "specified transport or endpoint configurations.  "
+                "Either this parameter or the "
+                "'--endpoint' parameter MUST be specified. Default: false."
+            ),
+        )
         parser.add_argument(
             "-it",
             "--inbound-transport",
@@ -1381,28 +1396,33 @@ def add_arguments(self, parser: ArgumentParser):
     def get_settings(self, args: Namespace):
         """Extract transport settings."""
         settings = {}
-        if args.inbound_transports:
-            settings["transport.inbound_configs"] = args.inbound_transports
-        else:
-            raise ArgsParseError("-it/--inbound-transport is required")
-        if args.outbound_transports:
-            settings["transport.outbound_configs"] = args.outbound_transports
+        if args.no_transport:
+            settings["transport.disabled"] = True
         else:
-            raise ArgsParseError("-ot/--outbound-transport is required")
-        settings["transport.enable_undelivered_queue"] = args.enable_undelivered_queue
+            if args.inbound_transports:
+                settings["transport.inbound_configs"] = args.inbound_transports
+            else:
+                raise ArgsParseError("-it/--inbound-transport is required")
+            if args.outbound_transports:
+                settings["transport.outbound_configs"] = args.outbound_transports
+            else:
+                raise ArgsParseError("-ot/--outbound-transport is required")
+            settings["transport.enable_undelivered_queue"] = (
+                args.enable_undelivered_queue
+            )
+            if args.max_message_size:
+                settings["transport.max_message_size"] = args.max_message_size
+            if args.max_outbound_retry:
+                settings["transport.max_outbound_retry"] = args.max_outbound_retry
+            if args.ws_heartbeat_interval:
+                settings["transport.ws.heartbeat_interval"] = args.ws_heartbeat_interval
+            if args.ws_timeout_interval:
+                settings["transport.ws.timeout_interval"] = args.ws_timeout_interval
 
         if args.label:
             settings["default_label"] = args.label
         if args.image_url:
             settings["image_url"] = args.image_url
-        if args.max_message_size:
-            settings["transport.max_message_size"] = args.max_message_size
-        if args.max_outbound_retry:
-            settings["transport.max_outbound_retry"] = args.max_outbound_retry
-        if args.ws_heartbeat_interval:
-            settings["transport.ws.heartbeat_interval"] = args.ws_heartbeat_interval
-        if args.ws_timeout_interval:
-            settings["transport.ws.timeout_interval"] = args.ws_timeout_interval
 
         return settings
 

aries_cloudagent/config/default_context.py

@@ -116,11 +116,15 @@ async def load_plugins(self, context: InjectionContext):
         context.injector.bind_instance(PluginRegistry, plugin_registry)
 
         # Register standard protocol plugins
-        plugin_registry.register_package("aries_cloudagent.protocols")
+        if not self.settings.get("transport.disabled"):
+            plugin_registry.register_package("aries_cloudagent.protocols")
 
         # Currently providing admin routes only
         plugin_registry.register_plugin("aries_cloudagent.holder")
-        plugin_registry.register_plugin("aries_cloudagent.ledger")
+
+        if not self.settings.get("ledger.disabled"):
+            plugin_registry.register_package("aries_cloudagent.ledger")
+
         plugin_registry.register_plugin("aries_cloudagent.messaging.jsonld")
         plugin_registry.register_plugin("aries_cloudagent.resolver")
         plugin_registry.register_plugin("aries_cloudagent.settings")

aries_cloudagent/config/logging.py

@@ -324,45 +324,47 @@ def print_banner(
             # Title
             banner.title(agent_label or "ACA")
             # Inbound transports
-            banner.subtitle("Inbound Transports")
-            internal_in_transports = [
-                f"{transport.scheme}://{transport.host}:{transport.port}"
-                for transport in inbound_transports.values()
-                if not transport.is_external
-            ]
-            if internal_in_transports:
-                banner.list(internal_in_transports)
-            external_in_transports = [
-                f"{transport.scheme}://{transport.host}:{transport.port}"
-                for transport in inbound_transports.values()
-                if transport.is_external
-            ]
-            if external_in_transports:
-                banner.subtitle("  External Plugin")
-                banner.list(external_in_transports)
+            if inbound_transports:
+                banner.subtitle("Inbound Transports")
+                internal_in_transports = [
+                    f"{transport.scheme}://{transport.host}:{transport.port}"
+                    for transport in inbound_transports.values()
+                    if not transport.is_external
+                ]
+                if internal_in_transports:
+                    banner.list(internal_in_transports)
+                external_in_transports = [
+                    f"{transport.scheme}://{transport.host}:{transport.port}"
+                    for transport in inbound_transports.values()
+                    if transport.is_external
+                ]
+                if external_in_transports:
+                    banner.subtitle("  External Plugin")
+                    banner.list(external_in_transports)
 
             # Outbound transports
-            banner.subtitle("Outbound Transports")
-            internal_schemes = set().union(
-                *(
-                    transport.schemes
-                    for transport in outbound_transports.values()
-                    if not transport.is_external
+            if outbound_transports:
+                banner.subtitle("Outbound Transports")
+                internal_schemes = set().union(
+                    *(
+                        transport.schemes
+                        for transport in outbound_transports.values()
+                        if not transport.is_external
+                    )
                 )
-            )
-            if internal_schemes:
-                banner.list([f"{scheme}" for scheme in sorted(internal_schemes)])
-
-            external_schemes = set().union(
-                *(
-                    transport.schemes
-                    for transport in outbound_transports.values()
-                    if transport.is_external
+                if internal_schemes:
+                    banner.list([f"{scheme}" for scheme in sorted(internal_schemes)])
+
+                external_schemes = set().union(
+                    *(
+                        transport.schemes
+                        for transport in outbound_transports.values()
+                        if transport.is_external
+                    )
                 )
-            )
-            if external_schemes:
-                banner.subtitle("  External Plugin")
-                banner.list([f"{scheme}" for scheme in sorted(external_schemes)])
+                if external_schemes:
+                    banner.subtitle("  External Plugin")
+                    banner.list([f"{scheme}" for scheme in sorted(external_schemes)])
 
             # DID info
             if public_did:
@@ -387,36 +389,42 @@ def print_banner(
     def print_notices(cls, settings: Settings):
         """Print notices and warnings."""
         with Banner(border=":", length=80, file=sys.stderr) as banner:
-            banner.centered("⚠ DEPRECATION NOTICE: ⚠")
-            banner.hr()
             if settings.get("wallet.type", "in_memory").lower() == "indy":
+                banner.centered("⚠ DEPRECATION NOTICE: ⚠")
+                banner.hr()
                 banner.print(
                     "The Indy wallet type is deprecated, use Askar instead; see: "
                     "https://aca-py.org/main/deploying/IndySDKtoAskarMigration/",
                 )
                 banner.hr()
-            banner.print(
-                "Receiving a core DIDComm protocol with the "
-                "`did:sov:BzCbsNYhMrjHiqZDTUASHg;spec` prefix is deprecated. All parties "
-                "sending this prefix should be notified that support for receiving such "
-                "messages will be removed in a future release. "
-                "Use https://didcomm.org/ instead."
-            )
-            banner.hr()
-            banner.print(
-                "Aries RFC 0160: Connection Protocol is deprecated and support will be "
-                "removed in a future release; use RFC 0023: DID Exchange instead."
-            )
-            banner.hr()
-            banner.print(
-                "Aries RFC 0036: Issue Credential 1.0 is deprecated and support will be "
-                "removed in a future release; use RFC 0453: Issue Credential 2.0 instead."
-            )
-            banner.hr()
-            banner.print(
-                "Aries RFC 0037: Present Proof 1.0 is deprecated and support will be "
-                "removed in a future release; use RFC 0454: Present Proof 2.0 instead."
-            )
+            if not settings.get("transport.disabled"):
+                banner.centered("⚠ DEPRECATION NOTICE: ⚠")
+                banner.hr()
+                banner.print(
+                    "Receiving a core DIDComm protocol with the "
+                    "`did:sov:BzCbsNYhMrjHiqZDTUASHg;spec` prefix is deprecated. "
+                    "All parties sending this prefix should be notified that support "
+                    "for receiving such messages will be removed in a future release. "
+                    "Use https://didcomm.org/ instead."
+                )
+                banner.hr()
+                banner.print(
+                    "Aries RFC 0160: Connection Protocol is deprecated and "
+                    "support will be removed in a future release; "
+                    "use RFC 0023: DID Exchange instead."
+                )
+                banner.hr()
+                banner.print(
+                    "Aries RFC 0036: Issue Credential 1.0 is deprecated "
+                    "and support will be removed in a future release; "
+                    "use RFC 0453: Issue Credential 2.0 instead."
+                )
+                banner.hr()
+                banner.print(
+                    "Aries RFC 0037: Present Proof 1.0 is deprecated "
+                    "and support will be removed in a future release; "
+                    "use RFC 0454: Present Proof 2.0 instead."
+                )
         print()
 
 

aries_cloudagent/core/conductor.py

@@ -191,20 +191,22 @@ async def setup(self):
         ):
             LOGGER.warning("No ledger configured")
 
-        # Register all inbound transports
-        self.inbound_transport_manager = InboundTransportManager(
-            self.root_profile, self.inbound_message_router, self.handle_not_returned
-        )
-        await self.inbound_transport_manager.setup()
-        context.injector.bind_instance(
-            InboundTransportManager, self.inbound_transport_manager
-        )
+        if not context.settings.get("transport.disabled"):
+            # Register all inbound transports if enabled
+            self.inbound_transport_manager = InboundTransportManager(
+                self.root_profile, self.inbound_message_router, self.handle_not_returned
+            )
+            await self.inbound_transport_manager.setup()
+            context.injector.bind_instance(
+                InboundTransportManager, self.inbound_transport_manager
+            )
 
-        # Register all outbound transports
-        self.outbound_transport_manager = OutboundTransportManager(
-            self.root_profile, self.handle_not_delivered
-        )
-        await self.outbound_transport_manager.setup()
+        if not context.settings.get("transport.disabled"):
+            # Register all outbound transports
+            self.outbound_transport_manager = OutboundTransportManager(
+                self.root_profile, self.handle_not_delivered
+            )
+            await self.outbound_transport_manager.setup()
 
         # Initialize dispatcher
         self.dispatcher = Dispatcher(self.root_profile)
@@ -286,17 +288,18 @@ async def start(self) -> None:
         context = self.root_profile.context
         await self.check_for_valid_wallet_type(self.root_profile)
 
-        # Start up transports
-        try:
-            await self.inbound_transport_manager.start()
-        except Exception:
-            LOGGER.exception("Unable to start inbound transports")
-            raise
-        try:
-            await self.outbound_transport_manager.start()
-        except Exception:
-            LOGGER.exception("Unable to start outbound transports")
-            raise
+        if not context.settings.get("transport.disabled"):
+            # Start up transports if enabled
+            try:
+                await self.inbound_transport_manager.start()
+            except Exception:
+                LOGGER.exception("Unable to start inbound transports")
+                raise
+            try:
+                await self.outbound_transport_manager.start()
+            except Exception:
+                LOGGER.exception("Unable to start outbound transports")
+                raise
 
         # Start up Admin server
         if self.admin_server:
@@ -316,14 +319,23 @@ async def start(self) -> None:
         # Get agent label
         default_label = context.settings.get("default_label")
 
-        # Show some details about the configuration to the user
-        LoggingConfigurator.print_banner(
-            default_label,
-            self.inbound_transport_manager.registered_transports,
-            self.outbound_transport_manager.registered_transports,
-            self.setup_public_did and self.setup_public_did.did,
-            self.admin_server,
-        )
+        if context.settings.get("transport.disabled"):
+            LoggingConfigurator.print_banner(
+                default_label,
+                None,
+                None,
+                self.setup_public_did and self.setup_public_did.did,
+                self.admin_server,
+            )
+        else:
+            LoggingConfigurator.print_banner(
+                default_label,
+                self.inbound_transport_manager.registered_transports,
+                self.outbound_transport_manager.registered_transports,
+                self.setup_public_did and self.setup_public_did.did,
+                self.admin_server,
+            )
+
         LoggingConfigurator.print_notices(context.settings)
 
         # record ACA-Py version in Wallet, if needed
@@ -620,19 +632,24 @@ def dispatch_complete(self, message: InboundMessage, completed: CompletedTask):
     async def get_stats(self) -> dict:
         """Get the current stats tracked by the conductor."""
         stats = {
-            "in_sessions": len(self.inbound_transport_manager.sessions),
+            "in_sessions": (
+                len(self.inbound_transport_manager.sessions)
+                if self.inbound_transport_manager
+                else 0
+            ),
             "out_encode": 0,
             "out_deliver": 0,
             "task_active": self.dispatcher.task_queue.current_active,
             "task_done": self.dispatcher.task_queue.total_done,
             "task_failed": self.dispatcher.task_queue.total_failed,
             "task_pending": self.dispatcher.task_queue.current_pending,
         }
-        for m in self.outbound_transport_manager.outbound_buffer:
-            if m.state == QueuedOutboundMessage.STATE_ENCODE:
-                stats["out_encode"] += 1
-            if m.state == QueuedOutboundMessage.STATE_DELIVER:
-                stats["out_deliver"] += 1
+        if self.outbound_transport_manager:
+            for m in self.outbound_transport_manager.outbound_buffer:
+                if m.state == QueuedOutboundMessage.STATE_ENCODE:
+                    stats["out_encode"] += 1
+                if m.state == QueuedOutboundMessage.STATE_DELIVER:
+                    stats["out_deliver"] += 1
         return stats
 
     async def outbound_message_router(