diff --git a/aries_cloudagent/admin/server.py b/aries_cloudagent/admin/server.py
index 9230536808..4497a86ac1 100644
--- a/aries_cloudagent/admin/server.py
+++ b/aries_cloudagent/admin/server.py
@@ -2,9 +2,14 @@
 
 import asyncio
 import logging
-from typing import Callable, Coroutine, Sequence, Set
+import re
 import uuid
 
+from typing import Callable, Coroutine, Sequence, Set
+
+import aiohttp_cors
+import jwt
+
 from aiohttp import web
 from aiohttp_apispec import (
     docs,
@@ -12,8 +17,6 @@
     setup_aiohttp_apispec,
     validation_middleware,
 )
-import aiohttp_cors
-import jwt
 
 from marshmallow import fields
 
@@ -364,12 +367,20 @@ async def setup_context(request: web.Request, handler):
 
         middlewares.append(setup_context)
 
-        app = web.Application(middlewares=middlewares)
+        app = web.Application(
+            middlewares=middlewares,
+            client_max_size=(
+                self.context.settings.get("admin.admin_client_max_request_size", 1)
+                * 1024
+                * 1024
+            ),
+        )
 
         server_routes = [
             web.get("/", self.redirect_handler, allow_head=False),
             web.get("/plugins", self.plugins_handler, allow_head=False),
             web.get("/status", self.status_handler, allow_head=False),
+            web.get("/status/config", self.config_handler, allow_head=False),
             web.post("/status/reset", self.status_reset_handler),
             web.get("/status/live", self.liveliness_handler, allow_head=False),
             web.get("/status/ready", self.readiness_handler, allow_head=False),
@@ -523,6 +534,41 @@ async def plugins_handler(self, request: web.BaseRequest):
         plugins = registry and sorted(registry.plugin_names) or []
         return web.json_response({"result": plugins})
 
+    @docs(tags=["server"], summary="Fetch the server configuration")
+    @response_schema(AdminStatusSchema(), 200, description="")
+    async def config_handler(self, request: web.BaseRequest):
+        """
+        Request handler for the server configuration.
+
+        Args:
+            request: aiohttp request object
+
+        Returns:
+            The web response
+
+        """
+        config = {
+            k: self.context.settings[k]
+            for k in self.context.settings
+            if k
+            not in [
+                "admin.admin_api_key",
+                "multitenant.jwt_secret",
+                "wallet.key",
+                "wallet.rekey",
+                "wallet.seed",
+                "wallet.storage.creds",
+            ]
+        }
+        for index in range(len(config.get("admin.webhook_urls", []))):
+            config["admin.webhook_urls"][index] = re.sub(
+                r"#.*",
+                "",
+                config["admin.webhook_urls"][index],
+            )
+
+        return web.json_response(config)
+
     @docs(tags=["server"], summary="Fetch the server status")
     @response_schema(AdminStatusSchema(), 200, description="")
     async def status_handler(self, request: web.BaseRequest):
diff --git a/aries_cloudagent/admin/tests/test_admin_server.py b/aries_cloudagent/admin/tests/test_admin_server.py
index d6137f917f..52c3cbf94c 100644
--- a/aries_cloudagent/admin/tests/test_admin_server.py
+++ b/aries_cloudagent/admin/tests/test_admin_server.py
@@ -1,3 +1,5 @@
+import json
+
 from aiohttp import ClientSession, DummyCookieJar, TCPConnector, web
 from aiohttp.test_utils import unused_port
 
@@ -190,10 +192,12 @@ async def test_start_stop(self):
 
         settings = {
             "admin.admin_insecure_mode": False,
+            "admin.admin_client_max_request_size": 4,
             "admin.admin_api_key": "test-api-key",
         }
         server = self.get_admin_server(settings)
         await server.start()
+        assert server.app._client_max_size == 4 * 1024 * 1024
         with async_mock.patch.object(
             server, "websocket_queues", async_mock.MagicMock()
         ) as mock_wsq:
@@ -372,6 +376,42 @@ async def test_visit_secure_mode(self):
 
         await server.stop()
 
+    async def test_query_config(self):
+        settings = {
+            "admin.admin_insecure_mode": False,
+            "admin.admin_api_key": "test-api-key",
+            "admin.webhook_urls": ["localhost:8123/abc#secret", "localhost:8123/def"],
+            "multitenant.jwt_secret": "abc123",
+            "wallet.key": "abc123",
+            "wallet.rekey": "def456",
+            "wallet.seed": "00000000000000000000000000000000",
+            "wallet.storage.creds": "secret",
+        }
+        server = self.get_admin_server(settings)
+        await server.start()
+
+        async with self.client_session.get(
+            f"http://127.0.0.1:{self.port}/status/config",
+            headers={"x-api-key": "test-api-key"},
+        ) as response:
+            result = json.loads(await response.text())
+            assert "admin.admin_insecure_mode" in result
+            assert all(
+                k not in result
+                for k in [
+                    "admin.admin_api_key",
+                    "multitenant.jwt_secret",
+                    "wallet.key",
+                    "wallet.rekey",
+                    "wallet.seed",
+                    "wallet.storage.creds",
+                ]
+            )
+            assert result["admin.webhook_urls"] == [
+                "localhost:8123/abc",
+                "localhost:8123/def",
+            ]
+
     async def test_visit_shutting_down(self):
         settings = {
             "admin.admin_insecure_mode": True,
diff --git a/aries_cloudagent/config/argparse.py b/aries_cloudagent/config/argparse.py
index 900ec6c6ac..ae85ab0bb6 100644
--- a/aries_cloudagent/config/argparse.py
+++ b/aries_cloudagent/config/argparse.py
@@ -99,54 +99,73 @@ def add_arguments(self, parser: ArgumentParser):
             nargs=2,
             metavar=("<host>", "<port>"),
             env_var="ACAPY_ADMIN",
-            help="Specify the host and port on which to run the administrative server.\
-            If not provided, no admin server is made available.",
+            help=(
+                "Specify the host and port on which to run the administrative server. "
+                "If not provided, no admin server is made available."
+            ),
         )
         parser.add_argument(
             "--admin-api-key",
             type=str,
             metavar="<api-key>",
             env_var="ACAPY_ADMIN_API_KEY",
-            help="Protect all admin endpoints with the provided API key.\
-            API clients (e.g. the controller) must pass the key in the HTTP\
-            header using 'X-API-Key: <api key>'. Either this parameter or the\
-            '--admin-insecure-mode' parameter MUST be specified.",
+            help=(
+                "Protect all admin endpoints with the provided API key. "
+                "API clients (e.g. the controller) must pass the key in the HTTP "
+                "header using 'X-API-Key: <api key>'. Either this parameter or the "
+                "'--admin-insecure-mode' parameter MUST be specified."
+            ),
         )
         parser.add_argument(
             "--admin-insecure-mode",
             action="store_true",
             env_var="ACAPY_ADMIN_INSECURE_MODE",
-            help="Run the admin web server in insecure mode. DO NOT USE FOR\
-            PRODUCTION DEPLOYMENTS. The admin server will be publicly available\
-            to anyone who has access to the interface. Either this parameter or\
-            the '--api-key' parameter MUST be specified.",
+            help=(
+                "Run the admin web server in insecure mode. DO NOT USE FOR "
+                "PRODUCTION DEPLOYMENTS. The admin server will be publicly available "
+                "to anyone who has access to the interface. Either this parameter or "
+                "the '--api-key' parameter MUST be specified."
+            ),
         )
         parser.add_argument(
             "--no-receive-invites",
             action="store_true",
             env_var="ACAPY_NO_RECEIVE_INVITES",
-            help="Prevents an agent from receiving invites by removing the\
-            '/connections/receive-invite' route from the administrative\
-            interface. Default: false.",
+            help=(
+                "Prevents an agent from receiving invites by removing the "
+                "'/connections/receive-invite' route from the administrative "
+                "interface. Default: false."
+            ),
         )
         parser.add_argument(
             "--help-link",
             type=str,
             metavar="<help-url>",
             env_var="ACAPY_HELP_LINK",
-            help="A URL to an administrative interface help web page that a controller\
-            user interface can get from the agent and provide as a link to users.",
+            help=(
+                "A URL to an administrative interface help web page that a controller "
+                "user interface can get from the agent and provide as a link to users."
+            ),
         )
         parser.add_argument(
             "--webhook-url",
             action="append",
             metavar="<url#api_key>",
             env_var="ACAPY_WEBHOOK_URL",
-            help="Send webhooks containing internal state changes to the specified\
-            URL. Optional API key to be passed in the request body can be appended using\
-            a hash separator [#]. This is useful for a controller to monitor agent events\
-            and respond to those events using the admin API. If not specified, \
-            webhooks are not published by the agent.",
+            help=(
+                "Send webhooks containing internal state changes to the specified "
+                "URL. Optional API key to be passed in the request body can be "
+                "appended using a hash separator [#]. This is useful for a controller "
+                "to monitor agent events and respond to those events using the "
+                "admin API. If not specified, webhooks are not published by the agent."
+            ),
+        )
+        parser.add_argument(
+            "--admin-client-max-request-size",
+            default=1,
+            type=BoundedInt(min=1, max=16),
+            env_var="ACAPY_ADMIN_CLIENT_MAX_REQUEST_SIZE",
+            help="Maximum client request size to admin server, in megabytes: default 1",
         )
 
     def get_settings(self, args: Namespace):
@@ -161,7 +180,7 @@ def get_settings(self, args: Namespace):
             ):
                 raise ArgsParseError(
                     "Either --admin-api-key or --admin-insecure-mode "
-                    + "must be set but not both."
+                    "must be set but not both."
                 )
 
             settings["admin.admin_api_key"] = admin_api_key
@@ -179,6 +198,10 @@ def get_settings(self, args: Namespace):
             if hook_url:
                 hook_urls.append(hook_url)
             settings["admin.webhook_urls"] = hook_urls
+
+            settings["admin.admin_client_max_request_size"] = (
+                args.admin_client_max_request_size or 1
+            )
         return settings
 
 
@@ -194,9 +217,11 @@ def add_arguments(self, parser: ArgumentParser):
             "--debug",
             action="store_true",
             env_var="ACAPY_DEBUG",
-            help="Enables a remote debugging service that can be accessed\
-            using ptvsd for Visual Studio Code. The framework will wait\
-            for the debugger to connect at start-up. Default: false.",
+            help=(
+                "Enables a remote debugging service that can be accessed "
+                "using ptvsd for Visual Studio Code. The framework will wait "
+                "for the debugger to connect at start-up. Default: false."
+            ),
         )
         parser.add_argument(
             "--debug-seed",
@@ -216,29 +241,37 @@ def add_arguments(self, parser: ArgumentParser):
             "--debug-credentials",
             action="store_true",
             env_var="ACAPY_DEBUG_CREDENTIALS",
-            help="Enable additional logging around credential exchanges.\
-            Default: false.",
+            help=(
+                "Enable additional logging around credential exchanges. "
+                "Default: false."
+            ),
         )
         parser.add_argument(
             "--debug-presentations",
             action="store_true",
             env_var="ACAPY_DEBUG_PRESENTATIONS",
-            help="Enable additional logging around presentation exchanges.\
-            Default: false.",
+            help=(
+                "Enable additional logging around presentation exchanges. "
+                "Default: false."
+            ),
         )
         parser.add_argument(
             "--invite",
             action="store_true",
             env_var="ACAPY_INVITE",
-            help="After startup, generate and print a new out-of-band connection invitation\
-            URL. Default: false.",
+            help=(
+                "After startup, generate and print a new out-of-band connection "
+                "invitation URL. Default: false."
+            ),
         )
         parser.add_argument(
             "--connections-invite",
             action="store_true",
             env_var="ACAPY_CONNECTIONS_INVITE",
-            help="After startup, generate and print a new connections protocol \
-            style invitation URL. Default: false.",
+            help=(
+                "After startup, generate and print a new connections protocol "
+                "style invitation URL. Default: false."
+            ),
         )
         parser.add_argument(
             "--invite-label",
@@ -279,36 +312,46 @@ def add_arguments(self, parser: ArgumentParser):
             "--auto-accept-invites",
             action="store_true",
             env_var="ACAPY_AUTO_ACCEPT_INVITES",
-            help="Automatically accept invites without firing a webhook event or\
-            waiting for an admin request. Default: false.",
+            help=(
+                "Automatically accept invites without firing a webhook event or "
+                "waiting for an admin request. Default: false."
+            ),
         )
         parser.add_argument(
             "--auto-accept-requests",
             action="store_true",
             env_var="ACAPY_AUTO_ACCEPT_REQUESTS",
-            help="Automatically accept connection requests without firing\
-            a webhook event or waiting for an admin request. Default: false.",
+            help=(
+                "Automatically accept connection requests without firing "
+                "a webhook event or waiting for an admin request. Default: false."
+            ),
         )
         parser.add_argument(
             "--auto-respond-messages",
             action="store_true",
             env_var="ACAPY_AUTO_RESPOND_MESSAGES",
-            help="Automatically respond to basic messages indicating the message was\
-            received. Default: false.",
+            help=(
+                "Automatically respond to basic messages indicating the message was "
+                "received. Default: false."
+            ),
         )
         parser.add_argument(
             "--auto-respond-credential-proposal",
             action="store_true",
             env_var="ACAPY_AUTO_RESPOND_CREDENTIAL_PROPOSAL",
-            help="Auto-respond to credential proposals with corresponding "
-            + "credential offers",
+            help=(
+                "Auto-respond to credential proposals with corresponding "
+                "credential offers"
+            ),
         )
         parser.add_argument(
             "--auto-respond-credential-offer",
             action="store_true",
             env_var="ACAPY_AUTO_RESPOND_CREDENTIAL_OFFER",
-            help="Automatically respond to Indy credential offers with a credential\
-            request. Default: false",
+            help=(
+                "Automatically respond to Indy credential offers with a credential "
+                "request. Default: false"
+            ),
         )
         parser.add_argument(
             "--auto-respond-credential-request",
@@ -320,30 +363,39 @@ def add_arguments(self, parser: ArgumentParser):
             "--auto-respond-presentation-proposal",
             action="store_true",
             env_var="ACAPY_AUTO_RESPOND_PRESENTATION_PROPOSAL",
-            help="Auto-respond to presentation proposals with corresponding "
-            + "presentation requests",
+            help=(
+                "Auto-respond to presentation proposals with corresponding "
+                "presentation requests"
+            ),
         )
         parser.add_argument(
             "--auto-respond-presentation-request",
             action="store_true",
             env_var="ACAPY_AUTO_RESPOND_PRESENTATION_REQUEST",
-            help="Automatically respond to Indy presentation requests with a\
-            constructed presentation if a corresponding credential can be retrieved\
-            for every referent in the presentation request. Default: false.",
+            help=(
+                "Automatically respond to Indy presentation requests with a "
+                "constructed presentation if a corresponding credential can be "
+                "retrieved for every referent in the presentation request. "
+                "Default: false."
+            ),
         )
         parser.add_argument(
             "--auto-store-credential",
             action="store_true",
             env_var="ACAPY_AUTO_STORE_CREDENTIAL",
-            help="Automatically store an issued credential upon receipt.\
-            Default: false.",
+            help=(
+                "Automatically store an issued credential upon receipt. "
+                "Default: false."
+            ),
         )
         parser.add_argument(
             "--auto-verify-presentation",
             action="store_true",
             env_var="ACAPY_AUTO_VERIFY_PRESENTATION",
-            help="Automatically verify a presentation when it is received.\
-            Default: false.",
+            help=(
+                "Automatically verify a presentation when it is received. "
+                "Default: false."
+            ),
         )
 
     def get_settings(self, args: Namespace) -> dict:
@@ -408,8 +460,10 @@ def add_arguments(self, parser: ArgumentParser):
         parser.add_argument(
             "--arg-file",
             is_config_file=True,
-            help="Load aca-py arguments from the specified file.  Note that\
-            this file *must* be in YAML format.",
+            help=(
+                "Load aca-py arguments from the specified file.  Note that "
+                "this file *must* be in YAML format."
+            ),
         )
         parser.add_argument(
             "--plugin",
@@ -419,19 +473,23 @@ def add_arguments(self, parser: ArgumentParser):
             required=False,
             metavar="<module>",
             env_var="ACAPY_PLUGIN",
-            help="Load <module> as external plugin module. Multiple\
-            instances of this parameter can be specified.",
+            help=(
+                "Load <module> as external plugin module. Multiple "
+                "instances of this parameter can be specified."
+            ),
         )
         parser.add_argument(
             "--storage-type",
             type=str,
             metavar="<storage-type>",
             env_var="ACAPY_STORAGE_TYPE",
-            help="Specifies the type of storage provider to use for the internal\
-            storage engine. This storage interface is used to store internal state.\
-            Supported internal storage types are 'basic' (memory)\
-            and 'indy'.  The default (if not specified) is 'indy' if the wallet type\
-            is set to 'indy', otherwise 'basic'.",
+            help=(
+                "Specifies the type of storage provider to use for the internal "
+                "storage engine. This storage interface is used to store internal "
+                "state  Supported internal storage types are 'basic' (memory) "
+                "and 'indy'.  The default (if not specified) is 'indy' if the "
+                "wallet type is set to 'indy', otherwise 'basic'."
+            ),
         )
         parser.add_argument(
             "-e",
@@ -440,15 +498,17 @@ def add_arguments(self, parser: ArgumentParser):
             nargs="+",
             metavar="<endpoint>",
             env_var="ACAPY_ENDPOINT",
-            help="Specifies the endpoints to put into DIDDocs\
-            to inform other agents of where they should send messages destined\
-            for this agent. Each endpoint could be one of the specified inbound\
-            transports for this agent, or the endpoint could be that of\
-            another agent (e.g. 'https://example.com/agent-endpoint') if the\
-            routing of messages to this agent by a mediator is configured.\
-            The first endpoint specified will be used in invitations.\
-            The endpoints are used in the formation of a connection\
-            with another agent.",
+            help=(
+                "Specifies the endpoints to put into DIDDocs "
+                "to inform other agents of where they should send messages destined "
+                "for this agent. Each endpoint could be one of the specified inbound "
+                "transports for this agent, or the endpoint could be that of "
+                "another agent (e.g. 'https://example.com/agent-endpoint') if the "
+                "routing of messages to this agent by a mediator is configured. "
+                "The first endpoint specified will be used in invitations. "
+                "The endpoints are used in the formation of a connection "
+                "with another agent."
+            ),
         )
         parser.add_argument(
             "--profile-endpoint",
@@ -461,8 +521,7 @@ def add_arguments(self, parser: ArgumentParser):
             "--read-only-ledger",
             action="store_true",
             env_var="ACAPY_READ_ONLY_LEDGER",
-            help="Sets ledger to read-only to prevent updates.\
-            Default: false.",
+            help="Sets ledger to read-only to prevent updates. Default: false.",
         )
         parser.add_argument(
             "--tails-server-base-url",
@@ -476,8 +535,10 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<tails-server-upload-url>",
             env_var="ACAPY_TAILS_SERVER_UPLOAD_URL",
-            help="Sets the base url of the tails server for upload, defaulting to the\
-            tails server base url.",
+            help=(
+                "Sets the base url of the tails server for upload, defaulting to the "
+                "tails server base url."
+            ),
         )
 
     def get_settings(self, args: Namespace) -> dict:
@@ -519,8 +580,10 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<ledger-pool-name>",
             env_var="ACAPY_LEDGER_POOL_NAME",
-            help="Specifies the name of the indy pool to be opened.\
-            This is useful if you have multiple pool configurations.",
+            help=(
+                "Specifies the name of the indy pool to be opened. "
+                "This is useful if you have multiple pool configurations."
+            ),
         )
         parser.add_argument(
             "--genesis-transactions",
@@ -528,9 +591,11 @@ def add_arguments(self, parser: ArgumentParser):
             dest="genesis_transactions",
             metavar="<genesis-transactions>",
             env_var="ACAPY_GENESIS_TRANSACTIONS",
-            help='Specifies the genesis transactions to use to connect to\
-            an Hyperledger Indy ledger. The transactions are provided as string\
-            of JSON e.g. \'{"reqSignature":{},"txn":{"data":{"d... <snip>\'',
+            help=(
+                "Specifies the genesis transactions to use to connect to "
+                "a Hyperledger Indy ledger. The transactions are provided as string "
+                'of JSON e.g. \'{"reqSignature":{},"txn":{"data":{"d... <snip>}}}\''
+            ),
         )
         parser.add_argument(
             "--genesis-file",
@@ -546,18 +611,23 @@ def add_arguments(self, parser: ArgumentParser):
             dest="genesis_url",
             metavar="<genesis-url>",
             env_var="ACAPY_GENESIS_URL",
-            help="Specifies the url from which to download the genesis\
-            transactions. For example, if you are using 'von-network',\
-            the URL might be 'http://localhost:9000/genesis'.\
-            Genesis transactions URLs are available for the Sovrin test/main networks.",
+            help=(
+                "Specifies the url from which to download the genesis "
+                "transactions. For example, if you are using 'von-network', "
+                "the URL might be 'http://localhost:9000/genesis'. "
+                "Genesis transactions URLs are available for the "
+                "Sovrin test/main networks."
+            ),
         )
         parser.add_argument(
             "--no-ledger",
             action="store_true",
             env_var="ACAPY_NO_LEDGER",
-            help="Specifies that aca-py will run with no ledger configured.\
-            This must be set if running in no-ledger mode.  Overrides any\
-            specified ledger or genesis configurations.  Default: false.",
+            help=(
+                "Specifies that aca-py will run with no ledger configured. "
+                "This must be set if running in no-ledger mode.  Overrides any "
+                "specified ledger or genesis configurations.  Default: false."
+            ),
         )
         parser.add_argument(
             "--ledger-keepalive",
@@ -582,8 +652,8 @@ def get_settings(self, args: Namespace) -> dict:
             else:
                 raise ArgsParseError(
                     "One of --genesis-url --genesis-file or --genesis-transactions "
-                    + "must be specified (unless --no-ledger is specified to "
-                    + "explicitly configure aca-py to run with no ledger)."
+                    "must be specified (unless --no-ledger is specified to "
+                    "explicitly configure aca-py to run with no ledger)."
                 )
             if args.ledger_pool_name:
                 settings["ledger.pool_name"] = args.ledger_pool_name
@@ -617,8 +687,10 @@ def add_arguments(self, parser: ArgumentParser):
             metavar="<log-file>",
             default=None,
             env_var="ACAPY_LOG_FILE",
-            help="Overrides the output destination for the root logger (as defined\
-            by the log config file) to the named <log-file>.",
+            help=(
+                "Overrides the output destination for the root logger (as defined "
+                "by the log config file) to the named <log-file>."
+            ),
         )
         parser.add_argument(
             "--log-level",
@@ -627,8 +699,10 @@ def add_arguments(self, parser: ArgumentParser):
             metavar="<log-level>",
             default=None,
             env_var="ACAPY_LOG_LEVEL",
-            help="Specifies a custom logging level as one of:\
-            ('debug', 'info', 'warning', 'error', 'critical')",
+            help=(
+                "Specifies a custom logging level as one of: "
+                "('debug', 'info', 'warning', 'error', 'critical')"
+            ),
         )
 
     def get_settings(self, args: Namespace) -> dict:
@@ -655,16 +729,20 @@ def add_arguments(self, parser: ArgumentParser):
             "--auto-ping-connection",
             action="store_true",
             env_var="ACAPY_AUTO_PING_CONNECTION",
-            help="Automatically send a trust ping immediately after a\
-            connection response is accepted. Some agents require this before\
-            marking a connection as 'active'. Default: false.",
+            help=(
+                "Automatically send a trust ping immediately after a "
+                "connection response is accepted. Some agents require this before "
+                "marking a connection as 'active'. Default: false."
+            ),
         )
         parser.add_argument(
             "--invite-base-url",
             type=str,
             metavar="<base-url>",
             env_var="ACAPY_INVITE_BASE_URL",
-            help="Base URL to use when formatting connection invitations in URL format.",
+            help=(
+                "Base URL to use when formatting connection invitations in URL format."
+            ),
         )
         parser.add_argument(
             "--monitor-ping",
@@ -676,8 +754,10 @@ def add_arguments(self, parser: ArgumentParser):
             "--public-invites",
             action="store_true",
             env_var="ACAPY_PUBLIC_INVITES",
-            help="Send invitations out, and receive connection requests,\
-            using the public DID for the agent. Default: false.",
+            help=(
+                "Send invitations out, and receive connection requests, "
+                "using the public DID for the agent. Default: false."
+            ),
         )
         parser.add_argument(
             "--timing",
@@ -729,24 +809,30 @@ def add_arguments(self, parser: ArgumentParser):
             "--emit-new-didcomm-prefix",
             action="store_true",
             env_var="ACAPY_EMIT_NEW_DIDCOMM_PREFIX",
-            help="Emit protocol messages with new DIDComm prefix; i.e.,\
-            'https://didcomm.org/' instead of (default) prefix\
-            'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/'.",
+            help=(
+                "Emit protocol messages with new DIDComm prefix; i.e., "
+                "'https://didcomm.org/' instead of (default) prefix "
+                "'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/'."
+            ),
         )
         parser.add_argument(
             "--emit-new-didcomm-mime-type",
             action="store_true",
             env_var="ACAPY_EMIT_NEW_DIDCOMM_MIME_TYPE",
-            help="Send packed agent messages with the DIDComm MIME type\
-            as of RFC 0044; i.e., 'application/didcomm-envelope-enc'\
-            instead of 'application/ssi-agent-wire'.",
+            help=(
+                "Send packed agent messages with the DIDComm MIME type "
+                "as of RFC 0044; i.e., 'application/didcomm-envelope-enc' "
+                "instead of 'application/ssi-agent-wire'."
+            ),
         )
         parser.add_argument(
             "--exch-use-unencrypted-tags",
             action="store_true",
             env_var="ACAPY_EXCH_USE_UNENCRYPTED_TAGS",
-            help="Store tags for exchange protocols (credential and presentation)\
-            using unencrypted rather than encrypted tags",
+            help=(
+                "Store tags for exchange protocols (credential and presentation) "
+                "using unencrypted rather than encrypted tags"
+            ),
         )
 
     def get_settings(self, args: Namespace) -> dict:
@@ -818,8 +904,10 @@ def add_arguments(self, parser: ArgumentParser):
             "--auto-provision",
             action="store_true",
             env_var="ACAPY_AUTO_PROVISION",
-            help="If the requested profile does not exist, initialize it with\
-            the given parameters.",
+            help=(
+                "If the requested profile does not exist, initialize it with "
+                "the given parameters."
+            ),
         )
 
     def get_settings(self, args: Namespace):
@@ -847,12 +935,14 @@ def add_arguments(self, parser: ArgumentParser):
             nargs=3,
             metavar=("<module>", "<host>", "<port>"),
             env_var="ACAPY_INBOUND_TRANSPORT",
-            help="REQUIRED. Defines the inbound transport(s) on which the agent\
-            listens for receiving messages from other agents. This parameter can\
-            be specified multiple times to create multiple interfaces.\
-            Built-in inbound transport types include 'http' and 'ws'.\
-            However, other transports can be loaded by specifying an absolute\
-            module path.",
+            help=(
+                "REQUIRED. Defines the inbound transport(s) on which the agent "
+                "listens for receiving messages from other agents. This parameter can "
+                "be specified multiple times to create multiple interfaces. "
+                "Built-in inbound transport types include 'http' and 'ws'. "
+                "However, other transports can be loaded by specifying an absolute "
+                "module path."
+            ),
         )
         parser.add_argument(
             "-ot",
@@ -862,10 +952,12 @@ def add_arguments(self, parser: ArgumentParser):
             action="append",
             metavar="<module>",
             env_var="ACAPY_OUTBOUND_TRANSPORT",
-            help="REQUIRED. Defines the outbound transport(s) on which the agent\
-            will send outgoing messages to other agents. This parameter can be passed\
-            multiple times to supoort multiple transport types. Supported outbound\
-            transport types are 'http' and 'ws'.",
+            help=(
+                "REQUIRED. Defines the outbound transport(s) on which the agent "
+                "will send outgoing messages to other agents. This parameter can be "
+                "passed multiple times to supoort multiple transport types. "
+                "Supported outbound transport types are 'http' and 'ws'."
+            ),
         )
         parser.add_argument(
             "-oq",
@@ -873,9 +965,10 @@ def add_arguments(self, parser: ArgumentParser):
             dest="outbound_queue",
             type=str,
             env_var="ACAPY_OUTBOUND_TRANSPORT_QUEUE",
-            help="""Defines connection details for outbound queue in a single
-                    connection string. For example, 'redis://127.0.0.1:6379'.
-                 """,
+            help=(
+                "Defines connection details for outbound queue in a single "
+                "connection string; e.g., 'redis://127.0.0.1:6379'."
+            ),
         )
         parser.add_argument(
             "-oqp",
@@ -883,10 +976,11 @@ def add_arguments(self, parser: ArgumentParser):
             dest="outbound_queue_prefix",
             type=str,
             env_var="ACAPY_OUTBOUND_TRANSPORT_QUEUE_PREFIX",
-            help="""Defines the prefix used to generate the queue key. The
-                    default is 'acapy', which generates a queue key as follows:
-                    'acapy.outbound_transport'.
-                 """,
+            help=(
+                "Defines the prefix used to generate the queue key. The "
+                "default is 'acapy', which generates a queue key as follows: "
+                "'acapy.outbound_transport'."
+            ),
         )
         parser.add_argument(
             "-oqc",
@@ -894,13 +988,15 @@ def add_arguments(self, parser: ArgumentParser):
             dest="outbound_queue_class",
             type=str,
             env_var="ACAPY_OUTBOUND_TRANSPORT_QUEUE_CLASS",
-            help="""Defines the location of the Outbound Queue Engine. This must be
-                    a 'dotpath' to a Python module on the PYTHONPATH, followed by a
-                    colon, followed by the name of a Python class that implements
-                    BaseOutboundQueue. This commandline option is the official entry
-                    point of ACA-py's pluggable queue interface. The default value is:
-                    'aries_cloudagent.transport.outbound.queue.redis:RedisOutboundQueue'
-                 """,
+            help=(
+                "Defines the location of the Outbound Queue Engine. This must be "
+                "a 'dotpath' to a Python module on the PYTHONPATH, followed by a "
+                "colon, followed by the name of a Python class that implements "
+                "BaseOutboundQueue. This commandline option is the official entry "
+                "point of ACA-py's pluggable queue interface. The default value is: "
+                "'aries_cloudagent.transport.outbound.queue.redis:RedisOutboundQueue'."
+                ""
+            ),
         )
         parser.add_argument(
             "-l",
@@ -908,15 +1004,19 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<label>",
             env_var="ACAPY_LABEL",
-            help="Specifies the label for this agent. This label is publicized\
-            (self-attested) to other agents as part of forming a connection.",
+            help=(
+                "Specifies the label for this agent. This label is publicized "
+                "(self-attested) to other agents as part of forming a connection."
+            ),
         )
         parser.add_argument(
             "--image-url",
             type=str,
             env_var="ACAPY_IMAGE_URL",
-            help="Specifies the image url for this agent. This image url is publicized\
-            (self-attested) to other agents as part of forming a connection.",
+            help=(
+                "Specifies the image url for this agent. This image url is publicized "
+                "(self-attested) to other agents as part of forming a connection."
+            ),
         )
         parser.add_argument(
             "--max-message-size",
@@ -930,18 +1030,22 @@ def add_arguments(self, parser: ArgumentParser):
             "--enable-undelivered-queue",
             action="store_true",
             env_var="ACAPY_ENABLE_UNDELIVERED_QUEUE",
-            help="Enable the outbound undelivered queue that enables this agent\
-            to hold messages for delivery to agents without an endpoint. This\
-            option will require additional memory to store messages in the queue.",
+            help=(
+                "Enable the outbound undelivered queue that enables this agent "
+                "to hold messages for delivery to agents without an endpoint. This "
+                "option will require additional memory to store messages in the queue."
+            ),
         )
         parser.add_argument(
             "--max-outbound-retry",
             default=4,
             type=BoundedInt(min=1),
             env_var="ACAPY_MAX_OUTBOUND_RETRY",
-            help="Set the maximum retry number for undelivered outbound\
-            messages. Increasing this number might cause to increase the\
-            accumulated messages in message queue. Default value is 4.",
+            help=(
+                "Set the maximum retry number for undelivered outbound "
+                "messages. Increasing this number might cause to increase the "
+                "accumulated messages in message queue. Default value is 4."
+            ),
         )
 
     def get_settings(self, args: Namespace):
@@ -997,25 +1101,33 @@ def add_arguments(self, parser: ArgumentParser):
         parser.add_argument(
             "--open-mediation",
             action="store_true",
-            help="Enables didcomm mediation. After establishing a connection, if enabled, \
-                an agent may request message mediation, which will allow the mediator to \
-                forward messages on behalf of the recipient. See aries-rfc:0211.",
+            env_var="ACAPY_MEDIATION_OPEN",
+            help=(
+                "Enables didcomm mediation. After establishing a connection, "
+                "if enabled, an agent may request message mediation, which will "
+                "allow the mediator to forward messages on behalf of the recipient. "
+                "See aries-rfc:0211."
+            ),
         )
         parser.add_argument(
             "--mediator-invitation",
             type=str,
             metavar="<invite URL to mediator>",
             env_var="ACAPY_MEDIATION_INVITATION",
-            help="Connect to mediator through provided invitation\
-            and send mediation request and set as default mediator.",
+            help=(
+                "Connect to mediator through provided invitation "
+                "and send mediation request and set as default mediator."
+            ),
         )
         parser.add_argument(
             "--mediator-connections-invite",
             action="store_true",
             env_var="ACAPY_MEDIATION_CONNECTIONS_INVITE",
-            help="Connect to mediator through a connection invitation. \
-                If not specified, connect using an OOB invitation. \
-                Default: false.",
+            help=(
+                "Connect to mediator through a connection invitation. "
+                "If not specified, connect using an OOB invitation. "
+                "Default: false."
+            ),
         )
         parser.add_argument(
             "--default-mediator-id",
@@ -1066,18 +1178,22 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<wallet-seed>",
             env_var="ACAPY_WALLET_SEED",
-            help="Specifies the seed to use for the creation of a public\
-            DID for the agent to use with a Hyperledger Indy ledger, or a local\
-            ('--wallet-local-did') DID. If public, the DID must already exist\
-            on the ledger.",
+            help=(
+                "Specifies the seed to use for the creation of a public "
+                "DID for the agent to use with a Hyperledger Indy ledger, or a local "
+                "('--wallet-local-did') DID. If public, the DID must already exist "
+                "on the ledger."
+            ),
         )
         parser.add_argument(
             "--wallet-local-did",
             action="store_true",
             env_var="ACAPY_WALLET_LOCAL_DID",
-            help="If this parameter is set, provisions the wallet with a\
-            local DID from the '--seed' parameter, instead of a public DID\
-            to use with a Hyperledger Indy ledger.",
+            help=(
+                "If this parameter is set, provisions the wallet with a "
+                "local DID from the '--seed' parameter, instead of a public DID "
+                "to use with a Hyperledger Indy ledger."
+            ),
         )
         parser.add_argument(
             "--wallet-key",
@@ -1091,16 +1207,20 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<wallet-rekey>",
             env_var="ACAPY_WALLET_REKEY",
-            help="Specifies a new master key value to which to rotate and to\
-            open the wallet next time.",
+            help=(
+                "Specifies a new master key value to which to rotate and to "
+                "open the wallet next time."
+            ),
         )
         parser.add_argument(
             "--wallet-name",
             type=str,
             metavar="<wallet-name>",
             env_var="ACAPY_WALLET_NAME",
-            help="Specifies the wallet name to be used by the agent.\
-            This is useful if your deployment has multiple wallets.",
+            help=(
+                "Specifies the wallet name to be used by the agent. "
+                "This is useful if your deployment has multiple wallets."
+            ),
         )
         parser.add_argument(
             "--wallet-type",
@@ -1108,9 +1228,11 @@ def add_arguments(self, parser: ArgumentParser):
             metavar="<wallet-type>",
             default="basic",
             env_var="ACAPY_WALLET_TYPE",
-            help="Specifies the type of Indy wallet provider to use.\
-            Supported internal storage types are 'basic' (memory) and 'indy'.\
-            The default (if not specified) is 'basic'.",
+            help=(
+                "Specifies the type of Indy wallet provider to use. "
+                "Supported internal storage types are 'basic' (memory) and 'indy'. "
+                "The default (if not specified) is 'basic'."
+            ),
         )
         parser.add_argument(
             "--wallet-storage-type",
@@ -1118,51 +1240,61 @@ def add_arguments(self, parser: ArgumentParser):
             metavar="<storage-type>",
             default="default",
             env_var="ACAPY_WALLET_STORAGE_TYPE",
-            help="Specifies the type of Indy wallet backend to use.\
-            Supported internal storage types are 'basic' (memory),\
-            'default' (sqlite), and 'postgres_storage'.  The default,\
-            if not specified, is 'default'.",
+            help=(
+                "Specifies the type of Indy wallet backend to use. "
+                "Supported internal storage types are 'basic' (memory), "
+                "'default' (sqlite), and 'postgres_storage'.  The default, "
+                "if not specified, is 'default'."
+            ),
         )
         parser.add_argument(
             "--wallet-storage-config",
             type=str,
             metavar="<storage-config>",
             env_var="ACAPY_WALLET_STORAGE_CONFIG",
-            help='Specifies the storage configuration to use for the wallet.\
-            This is required if you are for using \'postgres_storage\' wallet\
-            storage type. For example, \'{"url":"localhost:5432",\
-            "wallet_scheme":"MultiWalletSingleTable"}\'. This\
-            configuration maps to the indy sdk postgres plugin\
-            (PostgresConfig).',
+            help=(
+                "Specifies the storage configuration to use for the wallet. "
+                "This is required if you are for using 'postgres_storage' wallet "
+                'storage type. For example, \'{"url":"localhost:5432", '
+                '"wallet_scheme":"MultiWalletSingleTable"}\'. This '
+                "configuration maps to the indy sdk postgres plugin "
+                "(PostgresConfig)."
+            ),
         )
         parser.add_argument(
             "--wallet-storage-creds",
             type=str,
             metavar="<storage-creds>",
             env_var="ACAPY_WALLET_STORAGE_CREDS",
-            help='Specifies the storage credentials to use for the wallet.\
-            This is required if you are for using \'postgres_storage\' wallet\
-            For example, \'{"account":"postgres","password":\
-            "mysecretpassword","admin_account":"postgres",\
-            "admin_password":"mysecretpassword"}\'. This configuration maps\
-            to the indy sdk postgres plugin (PostgresCredentials). NOTE:\
-            admin_user must have the CREATEDB role or else initialization\
-            will fail.',
+            help=(
+                "Specifies the storage credentials to use for the wallet. "
+                "This is required if you are for using 'postgres_storage' wallet "
+                'For example, \'{"account":"postgres","password": '
+                '"mysecretpassword","admin_account":"postgres", '
+                '"admin_password":"mysecretpassword"}\'. This configuration maps '
+                "to the indy sdk postgres plugin (PostgresCredentials). NOTE: "
+                "admin_user must have the CREATEDB role or else initialization "
+                "will fail."
+            ),
         )
         parser.add_argument(
             "--replace-public-did",
             action="store_true",
             env_var="ACAPY_REPLACE_PUBLIC_DID",
-            help="If this parameter is set and an agent already has a public DID,\
-            and the '--seed' parameter specifies a new DID, the agent will use\
-            the new DID in place of the existing DID. Default: false.",
+            help=(
+                "If this parameter is set and an agent already has a public DID, "
+                "and the '--seed' parameter specifies a new DID, the agent will use "
+                "the new DID in place of the existing DID. Default: false."
+            ),
         )
         parser.add_argument(
             "--recreate-wallet",
             action="store_true",
             env_var="ACAPY_RECREATE_WALLET",
-            help="If an existing wallet exists with the same name, remove and\
-            recreate it during provisioning.",
+            help=(
+                "If an existing wallet exists with the same name, remove and "
+                "recreate it during provisioning."
+            ),
         )
 
     def get_settings(self, args: Namespace) -> dict:
@@ -1195,8 +1327,8 @@ def get_settings(self, args: Namespace) -> dict:
             # requires name, key
             if not args.wallet_name or not args.wallet_key:
                 raise ArgsParseError(
-                    "Parameters --wallet-name and --wallet-key must be provided"
-                    + " for indy wallets"
+                    "Parameters --wallet-name and --wallet-key must be provided "
+                    "for indy wallets"
                 )
             # postgres storage requires additional configuration
             if (
@@ -1205,8 +1337,8 @@ def get_settings(self, args: Namespace) -> dict:
             ):
                 if not args.wallet_storage_config or not args.wallet_storage_creds:
                     raise ArgsParseError(
-                        "Parameters --wallet-storage-config and --wallet-storage-creds"
-                        + " must be provided for indy postgres wallets"
+                        "Parameters --wallet-storage-config and --wallet-storage-creds "
+                        "must be provided for indy postgres wallets"
                     )
         return settings
 
@@ -1230,9 +1362,11 @@ def add_arguments(self, parser: ArgumentParser):
             type=str,
             metavar="<jwt-secret>",
             env_var="ACAPY_MULTITENANT_JWT_SECRET",
-            help="Specify the secret to be used for Json Web Token (JWT) \
-            creation and verification. The JWTs are used to authenticate and authorize \
-            multitenant wallets.",
+            help=(
+                "Specify the secret to be used for Json Web Token (JWT) creation and "
+                "verification. The JWTs are used to authenticate and authorize "
+                "multitenant wallets."
+            ),
         )
         parser.add_argument(
             "--multitenant-admin",
diff --git a/aries_cloudagent/config/default_context.py b/aries_cloudagent/config/default_context.py
index 631ad61677..7c213f5f3b 100644
--- a/aries_cloudagent/config/default_context.py
+++ b/aries_cloudagent/config/default_context.py
@@ -9,6 +9,8 @@
 from ..core.plugin_registry import PluginRegistry
 from ..core.profile import ProfileManager, ProfileManagerProvider
 from ..core.protocol_registry import ProtocolRegistry
+from ..resolver.did_resolver import DIDResolver
+from ..resolver.did_resolver_registry import DIDResolverRegistry
 from ..tails.base import BaseTailsServer
 from ..ledger.indy import IndySdkLedgerPool, IndySdkLedgerPoolProvider
 
@@ -17,7 +19,6 @@
 from ..protocols.didcomm_prefix import DIDCommPrefix
 from ..protocols.introduction.v0_1.base_service import BaseIntroductionService
 from ..protocols.introduction.v0_1.demo_service import DemoIntroductionService
-
 from ..transport.wire_format import BaseWireFormat
 from ..utils.stats import Collector
 
@@ -41,6 +42,13 @@ async def build_context(self) -> InjectionContext:
         # Global protocol registry
         context.injector.bind_instance(ProtocolRegistry, ProtocolRegistry())
 
+        # Global did resolver registry
+        did_resolver_registry = DIDResolverRegistry()
+        context.injector.bind_instance(DIDResolverRegistry, did_resolver_registry)
+
+        # Global did resolver
+        context.injector.bind_instance(DIDResolver, DIDResolver(did_resolver_registry))
+
         await self.bind_providers(context)
         await self.load_plugins(context)
 
@@ -103,7 +111,9 @@ async def load_plugins(self, context: InjectionContext):
             "aries_cloudagent.messaging.credential_definitions"
         )
         plugin_registry.register_plugin("aries_cloudagent.messaging.schemas")
+        # plugin_registry.register_plugin("aries_cloudagent.messaging.jsonld")
         plugin_registry.register_plugin("aries_cloudagent.revocation")
+        plugin_registry.register_plugin("aries_cloudagent.resolver")
         plugin_registry.register_plugin("aries_cloudagent.wallet")
 
         if context.settings.get("multitenant.admin_enabled"):
diff --git a/aries_cloudagent/messaging/decorators/attach_decorator.py b/aries_cloudagent/messaging/decorators/attach_decorator.py
index 205cb1b792..1fbb6dfe94 100644
--- a/aries_cloudagent/messaging/decorators/attach_decorator.py
+++ b/aries_cloudagent/messaging/decorators/attach_decorator.py
@@ -22,6 +22,8 @@
     set_urlsafe_b64,
     str_to_b64,
     unpad,
+    naked_to_did_key,
+    did_key_to_naked,
 )
 from ..models.base import BaseModel, BaseModelError, BaseModelSchema
 from ..valid import (
@@ -33,9 +35,6 @@
     UUIDFour,
 )
 
-MULTIBASE_B58_BTC = "z"
-MULTICODEC_ED25519_PUB = b"\xed"
-
 
 class AttachDecoratorDataJWSHeader(BaseModel):
     """Attach decorator data JWS header."""
@@ -198,19 +197,17 @@ def validate_single_xor_multi_sig(self, data: Mapping, **kwargs):
 def did_key(verkey: str) -> str:
     """Qualify verkey into DID key if need be."""
 
-    if verkey.startswith(f"did:key:{MULTIBASE_B58_BTC}"):
+    if verkey.startswith("did:key:"):
         return verkey
 
-    return f"did:key:{MULTIBASE_B58_BTC}" + bytes_to_b58(
-        MULTICODEC_ED25519_PUB + b58_to_bytes(verkey)
-    )
+    return naked_to_did_key(verkey)
 
 
 def raw_key(verkey: str) -> str:
     """Strip qualified key to raw key if need be."""
 
-    if verkey.startswith(f"did:key:{MULTIBASE_B58_BTC}"):
-        return bytes_to_b58(b58_to_bytes(verkey[9:])[1:])
+    if verkey.startswith("did:key:"):
+        return did_key_to_naked(verkey)
 
     return verkey
 
diff --git a/aries_cloudagent/messaging/jsonld/credential.py b/aries_cloudagent/messaging/jsonld/credential.py
index ca5ca3ea03..cbb0e5be0b 100644
--- a/aries_cloudagent/messaging/jsonld/credential.py
+++ b/aries_cloudagent/messaging/jsonld/credential.py
@@ -3,30 +3,21 @@
 import json
 
 from ...wallet.util import (
-    b58_to_bytes,
     b64_to_bytes,
     b64_to_str,
-    bytes_to_b58,
     bytes_to_b64,
     str_to_b64,
+    naked_to_did_key,
 )
 
 from .create_verify_data import create_verify_data
 from .error import BadJWSHeaderError
 
-MULTIBASE_B58_BTC = "z"
-MULTICODEC_ED25519_PUB = b"\xed"
-
 
 def did_key(verkey: str) -> str:
     """Qualify verkey into DID key if need be."""
 
-    return (
-        verkey
-        if verkey.startswith(f"did:key:{MULTIBASE_B58_BTC}")
-        else f"did:key:{MULTIBASE_B58_BTC}"
-        + bytes_to_b58(MULTICODEC_ED25519_PUB + b58_to_bytes(verkey))
-    )
+    return verkey if verkey.startswith("did:key:") else naked_to_did_key(verkey)
 
 
 def b64encode(str):
diff --git a/aries_cloudagent/messaging/jsonld/tests/test_credential.py b/aries_cloudagent/messaging/jsonld/tests/test_credential.py
index 14d8aae113..947596f0ba 100644
--- a/aries_cloudagent/messaging/jsonld/tests/test_credential.py
+++ b/aries_cloudagent/messaging/jsonld/tests/test_credential.py
@@ -10,7 +10,7 @@
 class TestCredential(AsyncTestCase):
     async def test_did_key(self):
         did_key = test_module.did_key(TEST_VERKEY)
-        assert did_key.startswith(f"did:key:{test_module.MULTIBASE_B58_BTC}")
+        assert did_key.startswith("did:key:z")
         assert did_key == test_module.did_key(did_key)
 
     async def test_verify_jws_header(self):
diff --git a/aries_cloudagent/protocols/coordinate_mediation/v1_0/tests/test_routes.py b/aries_cloudagent/protocols/coordinate_mediation/v1_0/tests/test_routes.py
index e1e655f321..fd9dad99a2 100644
--- a/aries_cloudagent/protocols/coordinate_mediation/v1_0/tests/test_routes.py
+++ b/aries_cloudagent/protocols/coordinate_mediation/v1_0/tests/test_routes.py
@@ -1,13 +1,11 @@
 import json
 
-import asynctest
-from asynctest import TestCase as AsyncTestCase
-from asynctest import mock as async_mock
-
-from aries_cloudagent.config.injection_context import InjectionContext
-from aries_cloudagent.messaging.request_context import RequestContext
+from asynctest import mock as async_mock, TestCase as AsyncTestCase
 
 from .....admin.request_context import AdminRequestContext
+from .....config.injection_context import InjectionContext
+from .....messaging.request_context import RequestContext
+
 from .. import routes as test_module
 from ..manager import MediationManager
 from ..models.mediation_record import MediationRecord
diff --git a/aries_cloudagent/protocols/present_proof/v1_0/manager.py b/aries_cloudagent/protocols/present_proof/v1_0/manager.py
index b28e7b5cf9..fbc4eb6755 100644
--- a/aries_cloudagent/protocols/present_proof/v1_0/manager.py
+++ b/aries_cloudagent/protocols/present_proof/v1_0/manager.py
@@ -279,10 +279,10 @@ async def create_presentation(
             if reft in req_attrs and reft in non_revoc_intervals:
                 requested_referents[reft]["non_revoked"] = non_revoc_intervals[reft]
 
-        preds_creds = requested_credentials.get("requested_predicates", {})
+        pred_creds = requested_credentials.get("requested_predicates", {})
         req_preds = presentation_request.get("requested_predicates", {})
-        for reft in preds_creds:
-            requested_referents[reft] = {"cred_id": preds_creds[reft]["cred_id"]}
+        for reft in pred_creds:
+            requested_referents[reft] = {"cred_id": pred_creds[reft]["cred_id"]}
             if reft in req_preds and reft in non_revoc_intervals:
                 requested_referents[reft]["non_revoked"] = non_revoc_intervals[reft]
 
@@ -294,6 +294,17 @@ async def create_presentation(
                     await holder.get_credential(credential_id)
                 )
 
+        # remove any timestamps that cannot correspond to non-revoc intervals
+        for r in ("requested_attributes", "requested_predicates"):
+            for reft, req_item in requested_credentials.get(r, {}).items():
+                if not credentials[req_item["cred_id"]].get(
+                    "rev_reg_id"
+                ) and req_item.pop("timestamp", None):
+                    LOGGER.info(
+                        f"Removed superfluous timestamp from requested_credentials {r} "
+                        f"{reft} for non-revocable credential {req_item['cred_id']}"
+                    )
+
         # Get all schemas, credential definitions, and revocation registries in use
         ledger = self._profile.inject(BaseLedger)
         schemas = {}
diff --git a/aries_cloudagent/protocols/present_proof/v1_0/tests/test_manager.py b/aries_cloudagent/protocols/present_proof/v1_0/tests/test_manager.py
index 62a1e2a12c..4a8c0da4fb 100644
--- a/aries_cloudagent/protocols/present_proof/v1_0/tests/test_manager.py
+++ b/aries_cloudagent/protocols/present_proof/v1_0/tests/test_manager.py
@@ -482,8 +482,9 @@ async def test_create_presentation_no_revocation(self):
             V10PresentationExchange, "save", autospec=True
         ) as save_ex, async_mock.patch.object(
             test_module, "AttachDecorator", autospec=True
-        ) as mock_attach_decorator:
-
+        ) as mock_attach_decorator, async_mock.patch.object(
+            test_module.LOGGER, "info", async_mock.MagicMock()
+        ) as mock_log_info:
             mock_attach_decorator.data_base64 = async_mock.MagicMock(
                 return_value=mock_attach_decorator
             )
@@ -498,6 +499,12 @@ async def test_create_presentation_no_revocation(self):
             save_ex.assert_called_once()
             assert exchange_out.state == V10PresentationExchange.STATE_PRESENTATION_SENT
 
+            # exercise superfluous timestamp removal
+            for pred_reft_spec in req_creds["requested_predicates"].values():
+                pred_reft_spec["timestamp"] = 1234567890
+            await self.manager.create_presentation(exchange_in, req_creds)
+            mock_log_info.assert_called_once()
+
     async def test_create_presentation_bad_revoc_state(self):
         exchange_in = V10PresentationExchange()
         indy_proof_req = await PRES_PREVIEW.indy_proof_request(
diff --git a/aries_cloudagent/protocols/present_proof/v2_0/manager.py b/aries_cloudagent/protocols/present_proof/v2_0/manager.py
index 24c0d6eedd..ef2bd8cde3 100644
--- a/aries_cloudagent/protocols/present_proof/v2_0/manager.py
+++ b/aries_cloudagent/protocols/present_proof/v2_0/manager.py
@@ -291,10 +291,10 @@ async def create_pres(
             if reft in req_attrs and reft in non_revoc_intervals:
                 requested_referents[reft]["non_revoked"] = non_revoc_intervals[reft]
 
-        preds_creds = requested_credentials.get("requested_predicates", {})
+        pred_creds = requested_credentials.get("requested_predicates", {})
         req_preds = proof_request.get("requested_predicates", {})
-        for reft in preds_creds:
-            requested_referents[reft] = {"cred_id": preds_creds[reft]["cred_id"]}
+        for reft in pred_creds:
+            requested_referents[reft] = {"cred_id": pred_creds[reft]["cred_id"]}
             if reft in req_preds and reft in non_revoc_intervals:
                 requested_referents[reft]["non_revoked"] = non_revoc_intervals[reft]
 
@@ -306,6 +306,17 @@ async def create_pres(
                     await holder.get_credential(credential_id)
                 )
 
+        # remove any timestamps that cannot correspond to non-revoc intervals
+        for r in ("requested_attributes", "requested_predicates"):
+            for reft, req_item in requested_credentials.get(r, {}).items():
+                if not credentials[req_item["cred_id"]].get(
+                    "rev_reg_id"
+                ) and req_item.pop("timestamp", None):
+                    LOGGER.info(
+                        f"Removed superfluous timestamp from requested_credentials {r} "
+                        f"{reft} for non-revocable credential {req_item['cred_id']}"
+                    )
+
         # Get all schemas, credential definitions, and revocation registries in use
         ledger = self._profile.inject(BaseLedger)
         schemas = {}
diff --git a/aries_cloudagent/protocols/present_proof/v2_0/tests/test_manager.py b/aries_cloudagent/protocols/present_proof/v2_0/tests/test_manager.py
index 4be874ba80..c4b29bdf33 100644
--- a/aries_cloudagent/protocols/present_proof/v2_0/tests/test_manager.py
+++ b/aries_cloudagent/protocols/present_proof/v2_0/tests/test_manager.py
@@ -524,8 +524,9 @@ async def test_create_pres_no_revocation(self):
             V20PresExRecord, "save", autospec=True
         ) as save_ex, async_mock.patch.object(
             test_module, "AttachDecorator", autospec=True
-        ) as mock_attach_decorator:
-
+        ) as mock_attach_decorator, async_mock.patch.object(
+            test_module.LOGGER, "info", async_mock.MagicMock()
+        ) as mock_log_info:
             mock_attach_decorator.data_base64 = async_mock.MagicMock(
                 return_value=mock_attach_decorator
             )
@@ -540,6 +541,12 @@ async def test_create_pres_no_revocation(self):
             save_ex.assert_called_once()
             assert px_rec_out.state == V20PresExRecord.STATE_PRESENTATION_SENT
 
+            # exercise superfluous timestamp removal
+            for pred_reft_spec in req_creds["requested_predicates"].values():
+                pred_reft_spec["timestamp"] = 1234567890
+            await self.manager.create_pres(px_rec_in, req_creds)
+            mock_log_info.assert_called_once()
+
     async def test_create_pres_bad_revoc_state(self):
         pres_request = V20PresRequest(
             formats=[
diff --git a/aries_cloudagent/protocols/problem_report/v1_0/tests/test_handler.py b/aries_cloudagent/protocols/problem_report/v1_0/tests/test_handler.py
index 559ca73390..ae3db794c8 100644
--- a/aries_cloudagent/protocols/problem_report/v1_0/tests/test_handler.py
+++ b/aries_cloudagent/protocols/problem_report/v1_0/tests/test_handler.py
@@ -1,9 +1,9 @@
 import pytest
 
-from aries_cloudagent.messaging.base_handler import HandlerException
-from aries_cloudagent.messaging.request_context import RequestContext
-from aries_cloudagent.messaging.responder import MockResponder
-from aries_cloudagent.transport.inbound.receipt import MessageReceipt
+from .....messaging.base_handler import HandlerException
+from .....messaging.request_context import RequestContext
+from .....messaging.responder import MockResponder
+from .....transport.inbound.receipt import MessageReceipt
 
 from ..handler import ProblemReportHandler
 from ..message import ProblemReport
diff --git a/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_handler.py b/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_handler.py
index 2a631a72be..9366908fb0 100644
--- a/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_handler.py
+++ b/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_handler.py
@@ -1,9 +1,9 @@
 import pytest
 
-from aries_cloudagent.messaging.base_handler import HandlerException
-from aries_cloudagent.messaging.request_context import RequestContext
-from aries_cloudagent.messaging.responder import MockResponder
-from aries_cloudagent.transport.inbound.receipt import MessageReceipt
+from ......messaging.base_handler import HandlerException
+from ......messaging.request_context import RequestContext
+from ......messaging.responder import MockResponder
+from ......transport.inbound.receipt import MessageReceipt
 
 from ...handlers.ping_handler import PingHandler
 from ...messages.ping import Ping
diff --git a/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_response_handler.py b/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_response_handler.py
index 32f4094523..f5b7cd13e4 100644
--- a/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_response_handler.py
+++ b/aries_cloudagent/protocols/trustping/v1_0/handlers/tests/test_ping_response_handler.py
@@ -1,9 +1,9 @@
 import pytest
 
-from aries_cloudagent.messaging.base_handler import HandlerException
-from aries_cloudagent.messaging.request_context import RequestContext
-from aries_cloudagent.messaging.responder import MockResponder
-from aries_cloudagent.transport.inbound.receipt import MessageReceipt
+from ......messaging.base_handler import HandlerException
+from ......messaging.request_context import RequestContext
+from ......messaging.responder import MockResponder
+from ......transport.inbound.receipt import MessageReceipt
 
 from ...handlers.ping_response_handler import PingResponseHandler
 from ...messages.ping_response import PingResponse
diff --git a/aries_cloudagent/resolver/__init__.py b/aries_cloudagent/resolver/__init__.py
new file mode 100644
index 0000000000..a697a06bcc
--- /dev/null
+++ b/aries_cloudagent/resolver/__init__.py
@@ -0,0 +1,27 @@
+"""Interfaces and base classes for DID Resolution."""
+
+import logging
+
+from ..config.injection_context import InjectionContext
+from ..config.provider import ClassProvider
+from .did_resolver_registry import DIDResolverRegistry
+
+LOGGER = logging.getLogger(__name__)
+
+
+async def setup(context: InjectionContext):
+    """Set up default resolvers."""
+    registry = context.inject(DIDResolverRegistry, required=False)
+    if not registry:
+        LOGGER.warning("No DID Resolver Registry instance found in context")
+        return
+
+    if context.settings.get("ledger.disabled"):
+        LOGGER.warning("Ledger is not configured, not loading IndyDIDResolver")
+        return
+
+    resolver = ClassProvider(
+        "aries_cloudagent.resolver.default.indy.IndyDIDResolver"
+    ).provide(context.settings, context.injector)
+    await resolver.setup(context)
+    registry.register(resolver)
diff --git a/aries_cloudagent/resolver/base.py b/aries_cloudagent/resolver/base.py
new file mode 100644
index 0000000000..0e2bfb1c5d
--- /dev/null
+++ b/aries_cloudagent/resolver/base.py
@@ -0,0 +1,79 @@
+"""Base Class for DID Resolvers."""
+
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Sequence, Union
+
+from pydid import DID, DIDDocument
+
+from ..config.injection_context import InjectionContext
+from ..core.error import BaseError
+from ..core.profile import Profile
+
+
+class ResolverError(BaseError):
+    """Base class for resolver exceptions."""
+
+
+class DIDNotFound(ResolverError):
+    """Raised when DID is not found in verifiable data registry."""
+
+
+class DIDMethodNotSupported(ResolverError):
+    """Raised when no resolver is registered for a given did method."""
+
+
+class ResolverType(Enum):
+    """Resolver Type declarations."""
+
+    NATIVE = "native"
+    NON_NATIVE = "non-native"
+
+
+class BaseDIDResolver(ABC):
+    """Base Class for DID Resolvers."""
+
+    def __init__(self, type_: ResolverType = None):
+        """Initialize BaseDIDResolver.
+
+        Args:
+            type_ (Type): Type of resolver, native or non-native
+        """
+        self.type = type_ or ResolverType.NON_NATIVE
+
+    @abstractmethod
+    async def setup(self, context: InjectionContext):
+        """Do asynchronous resolver setup."""
+
+    @property
+    def native(self):
+        """Return if this resolver is native."""
+        return self.type == ResolverType.NATIVE
+
+    @property
+    @abstractmethod
+    def supported_methods(self) -> Sequence[str]:
+        """Return list of DID methods supported by this resolver."""
+
+    def supports(self, method: str) -> bool:
+        """Return if this resolver supports the given method."""
+        return method in self.supported_methods
+
+    async def resolve(self, profile: Profile, did: Union[str, DID]) -> DIDDocument:
+        """Resolve a DID using this resolver."""
+        if isinstance(did, str):
+            did = DID(did)
+
+        if not self.supports(did.method):
+            raise DIDMethodNotSupported(
+                f"{self.__class__.__name__} does not support DID method {did.method}"
+            )
+
+        did = str(did)
+        did_document = await self._resolve(profile, did)
+        result = DIDDocument.deserialize(did_document)
+        return result
+
+    @abstractmethod
+    async def _resolve(self, profile: Profile, did: DID) -> dict:
+        """Resolve a DID using this resolver."""
diff --git a/aries_cloudagent/resolver/default/__init__.py b/aries_cloudagent/resolver/default/__init__.py
new file mode 100644
index 0000000000..467e43440c
--- /dev/null
+++ b/aries_cloudagent/resolver/default/__init__.py
@@ -0,0 +1 @@
+"""Resolvers included in ACA-Py by Default."""
diff --git a/aries_cloudagent/resolver/default/indy.py b/aries_cloudagent/resolver/default/indy.py
new file mode 100644
index 0000000000..ab24ca10ca
--- /dev/null
+++ b/aries_cloudagent/resolver/default/indy.py
@@ -0,0 +1,71 @@
+"""Indy DID Resolver.
+
+Resolution is performed using the IndyLedger class.
+"""
+from typing import Sequence
+
+from pydid import DID, DIDDocumentBuilder, VerificationSuite
+
+from ...config.injection_context import InjectionContext
+from ...core.profile import Profile
+from ...ledger.indy import IndySdkLedger
+from ...ledger.base import BaseLedger
+from ...ledger.error import LedgerError
+from ..base import BaseDIDResolver, DIDNotFound, ResolverError, ResolverType
+
+
+class NoIndyLedger(ResolverError):
+    """Raised when there is no indy ledger instance configured."""
+
+
+class IndyDIDResolver(BaseDIDResolver):
+    """Indy DID Resolver."""
+
+    VERIFICATION_METHOD_TYPE = "Ed25519VerificationKey2018"
+    AGENT_SERVICE_TYPE = "did-communication"
+    SUITE = VerificationSuite(VERIFICATION_METHOD_TYPE, "publicKeyBase58")
+
+    def __init__(self):
+        """Initialize Indy Resolver."""
+        super().__init__(ResolverType.NATIVE)
+
+    async def setup(self, context: InjectionContext):
+        """Perform required setup for Indy DID resolution."""
+
+    @property
+    def supported_methods(self) -> Sequence[str]:
+        """Return supported methods of Indy DID Resolver."""
+        return ["sov"]
+
+    async def _resolve(self, profile: Profile, did: str) -> dict:
+        """Resolve an indy DID."""
+        did = DID(did)
+        ledger = profile.inject(BaseLedger, required=False)
+        if not ledger or not isinstance(ledger, IndySdkLedger):
+            raise NoIndyLedger("No Indy ledger instance is configured.")
+
+        try:
+            async with ledger:
+                recipient_key = await ledger.get_key_for_did(did)
+                endpoint = await ledger.get_endpoint_for_did(did)
+        except LedgerError as err:
+            raise DIDNotFound(f"DID {did} could not be resolved") from err
+
+        builder = DIDDocumentBuilder(did)
+
+        vmethod = builder.verification_methods.add(
+            ident="key-1", suite=self.SUITE, material=recipient_key
+        )
+        builder.authentication.reference(vmethod.id)
+        builder.assertion_method.reference(vmethod.id)
+        if endpoint:
+            # TODO add priority
+            builder.services.add_didcomm(
+                ident=self.AGENT_SERVICE_TYPE,
+                type_=self.AGENT_SERVICE_TYPE,
+                endpoint=endpoint,
+                recipient_keys=[vmethod],
+                routing_keys=[],
+            )
+        result = builder.build()
+        return result.serialize()
diff --git a/aries_cloudagent/resolver/default/tests/__init__.py b/aries_cloudagent/resolver/default/tests/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/aries_cloudagent/resolver/default/tests/test_indy.py b/aries_cloudagent/resolver/default/tests/test_indy.py
new file mode 100644
index 0000000000..ce13d065d1
--- /dev/null
+++ b/aries_cloudagent/resolver/default/tests/test_indy.py
@@ -0,0 +1,70 @@
+"""Test IndyDIDResolver."""
+
+import pytest
+from asynctest import mock as async_mock
+
+from ....core.in_memory import InMemoryProfile
+from ....core.profile import Profile
+from ....ledger.base import BaseLedger
+from ....ledger.error import LedgerError
+from ...base import DIDNotFound, ResolverError
+from .. import indy as test_module
+from ..indy import IndyDIDResolver
+
+# pylint: disable=W0621
+TEST_DID0 = "did:sov:123"
+
+
+@pytest.fixture
+def resolver():
+    """Resolver fixture."""
+    yield IndyDIDResolver()
+
+
+@pytest.fixture
+def ledger():
+    """Ledger fixture."""
+    ledger = async_mock.MagicMock(spec=test_module.IndySdkLedger)
+    ledger.get_endpoint_for_did = async_mock.CoroutineMock(
+        return_value="https://github.com/"
+    )
+    ledger.get_key_for_did = async_mock.CoroutineMock(return_value="key")
+    yield ledger
+
+
+@pytest.fixture
+def profile(ledger):
+    """Profile fixture."""
+    profile = InMemoryProfile.test_profile()
+    profile.context.injector.bind_instance(BaseLedger, ledger)
+    yield profile
+
+
+def test_supported_methods(resolver: IndyDIDResolver):
+    """Test the supported_methods."""
+    assert resolver.supported_methods == ["sov"]
+    assert resolver.supports("sov")
+
+
+@pytest.mark.asyncio
+async def test_resolve(resolver: IndyDIDResolver, profile: Profile):
+    """Test resolve method."""
+    assert await resolver.resolve(profile, TEST_DID0)
+
+
+@pytest.mark.asyncio
+async def test_resolve_x_no_ledger(resolver: IndyDIDResolver, profile: Profile):
+    """Test resolve method with no ledger."""
+    profile.context.injector.clear_binding(BaseLedger)
+    with pytest.raises(ResolverError):
+        await resolver.resolve(profile, TEST_DID0)
+
+
+@pytest.mark.asyncio
+async def test_resolve_x_did_not_found(
+    resolver: IndyDIDResolver, ledger: BaseLedger, profile: Profile
+):
+    """Test resolve method when no did is found."""
+    ledger.get_key_for_did.side_effect = LedgerError
+    with pytest.raises(DIDNotFound):
+        await resolver.resolve(profile, TEST_DID0)
diff --git a/aries_cloudagent/resolver/did_resolver.py b/aries_cloudagent/resolver/did_resolver.py
new file mode 100644
index 0000000000..5c11d68c7b
--- /dev/null
+++ b/aries_cloudagent/resolver/did_resolver.py
@@ -0,0 +1,70 @@
+"""
+the did resolver.
+
+responsible for keeping track of all resolvers. more importantly
+retrieving did's from different sources provided by the method type.
+"""
+
+import logging
+from itertools import chain
+from typing import Union
+
+from pydid import DID, DIDUrl, DIDDocument, Service, VerificationMethod
+
+from ..core.profile import Profile
+from ..resolver.base import BaseDIDResolver, DIDMethodNotSupported, DIDNotFound
+from .did_resolver_registry import DIDResolverRegistry
+
+LOGGER = logging.getLogger(__name__)
+
+
+class DIDResolver:
+    """did resolver singleton."""
+
+    def __init__(self, registry: DIDResolverRegistry):
+        """Initialize a `didresolver` instance."""
+        self.did_resolver_registry = registry
+
+    async def resolve(self, profile: Profile, did: Union[str, DID]) -> DIDDocument:
+        """Retrieve did doc from public registry."""
+        # TODO Cache results
+        if isinstance(did, str):
+            did = DID(did)
+        for resolver in self._match_did_to_resolver(did):
+            try:
+                LOGGER.debug("Resolving DID %s with %s", did, resolver)
+                return await resolver.resolve(profile, did)
+            except DIDNotFound:
+                LOGGER.debug("DID %s not found by resolver %s", did, resolver)
+
+        raise DIDNotFound(f"DID {did} could not be resolved")
+
+    def _match_did_to_resolver(self, did: DID) -> BaseDIDResolver:
+        """Generate supported DID Resolvers.
+
+        Native resolvers are yielded first, in registered order followed by
+        non-native resolvers in registered order.
+        """
+        valid_resolvers = list(
+            filter(
+                lambda resolver: resolver.supports(did.method),
+                self.did_resolver_registry.resolvers,
+            )
+        )
+        native_resolvers = filter(lambda resolver: resolver.native, valid_resolvers)
+        non_native_resolvers = filter(
+            lambda resolver: not resolver.native, valid_resolvers
+        )
+        resolvers = list(chain(native_resolvers, non_native_resolvers))
+        if not resolvers:
+            raise DIDMethodNotSupported(f"DID method '{did.method}' not supported")
+        return resolvers
+
+    async def dereference(
+        self, profile: Profile, did_url: str
+    ) -> Union[Service, VerificationMethod]:
+        """Dereference a DID URL to its corresponding DID Doc object."""
+        # TODO Use cached DID Docs when possible
+        did_url = DIDUrl.parse(did_url)
+        doc = await self.resolve(profile, did_url.did)
+        return doc.dereference(did_url)
diff --git a/aries_cloudagent/resolver/did_resolver_registry.py b/aries_cloudagent/resolver/did_resolver_registry.py
new file mode 100644
index 0000000000..5db5d28ebf
--- /dev/null
+++ b/aries_cloudagent/resolver/did_resolver_registry.py
@@ -0,0 +1,26 @@
+"""In memmory storage for registering did resolvers."""
+
+import logging
+from typing import Sequence
+
+LOGGER = logging.getLogger(__name__)
+
+
+class DIDResolverRegistry:
+    """Registry for did resolvers."""
+
+    def __init__(self):
+        """Initialize list for did resolvers."""
+        self._resolvers = []
+
+    @property
+    def resolvers(
+        self,
+    ) -> Sequence[str]:
+        """Accessor for a list of all did resolvers."""
+        return self._resolvers
+
+    def register(self, resolver) -> None:
+        """Register a resolver."""
+        LOGGER.debug("Registering resolver %s", resolver)
+        self._resolvers.append(resolver)
diff --git a/aries_cloudagent/resolver/routes.py b/aries_cloudagent/resolver/routes.py
new file mode 100644
index 0000000000..59d451e029
--- /dev/null
+++ b/aries_cloudagent/resolver/routes.py
@@ -0,0 +1,143 @@
+"""
+Resolve did document admin routes.
+
+"/resolver/resolve/{did}": {
+    "get": {
+        "responses": {
+            "200": {
+                "schema": {
+                    "$ref": "#/definitions/DIDDoc"
+                },
+                "description": null
+            }
+        },
+        "parameters": [
+            {
+                "in": "path",
+                "name": "did",
+                "required": true,
+                "type": "string",
+                "pattern": "did:([a-z]+):((?:[a-zA-Z0-9._-]*:)*[a-zA-Z0-9._-]+)",
+                "description": "decentralize identifier(DID)",
+                "example": "did:ted:WgWxqztrNooG92RXvxSTWv"
+            }
+        ],
+        "tags": [ "resolver" ],
+        "summary": "Retrieve doc for requested did",
+        "produces": [ "application/json" ]
+    }
+}
+"""
+
+from aiohttp import web
+from aiohttp_apispec import docs, match_info_schema, response_schema
+from marshmallow import fields, validate
+
+from ..admin.request_context import AdminRequestContext
+from ..messaging.models.openapi import OpenAPISchema
+from .base import DIDMethodNotSupported, DIDNotFound, ResolverError
+from pydid.common import DID_PATTERN
+from .did_resolver import DIDResolver
+
+
+class W3cDIDDoc(validate.Regexp):
+    """Validate value against w3c DID document."""
+
+    EXAMPLE = "*"
+    PATTERN = DID_PATTERN
+
+    def __init__(self):
+        """Initializer."""
+
+        super().__init__(
+            W3cDIDDoc.PATTERN,
+            error="Value {input} is not a w3c decentralized identifier (DID) doc.",
+        )
+
+
+_W3cDIDDoc = {"validate": W3cDIDDoc(), "example": W3cDIDDoc.EXAMPLE}
+
+
+class DIDDocSchema(OpenAPISchema):
+    """Result schema for did document query."""
+
+    did_doc = fields.Str(
+        description="decentralize identifier(DID) document", required=True, **_W3cDIDDoc
+    )
+
+
+class W3cDID(validate.Regexp):
+    """Validate value against w3c DID."""
+
+    EXAMPLE = "did:ted:WgWxqztrNooG92RXvxSTWv"
+    PATTERN = DID_PATTERN
+
+    def __init__(self):
+        """Initializer."""
+
+        super().__init__(
+            W3cDID.PATTERN,
+            error="Value {input} is not a w3c decentralized identifier (DID)",
+        )
+
+
+_W3cDID = {"validate": W3cDID(), "example": W3cDID.EXAMPLE}
+
+
+class DIDMatchInfoSchema(OpenAPISchema):
+    """Path parameters and validators for request taking DID."""
+
+    did = fields.Str(
+        description="decentralize identifier(DID)", required=True, **_W3cDID
+    )
+
+
+@docs(tags=["resolver"], summary="Retrieve doc for requested did")
+@match_info_schema(DIDMatchInfoSchema())
+@response_schema(DIDDocSchema(), 200)
+async def resolve_did(request: web.BaseRequest):
+    """Retrieve a did document."""
+    context: AdminRequestContext = request["context"]
+
+    did = request.match_info["did"]
+    try:
+        session = await context.session()
+        resolver = session.inject(DIDResolver)
+        document = await resolver.resolve(context.profile, did)
+        result = document.serialize()
+    except DIDNotFound as err:
+        raise web.HTTPNotFound(reason=err.roll_up) from err
+    except DIDMethodNotSupported as err:
+        raise web.HTTPNotImplemented(reason=err.roll_up) from err
+    except ResolverError as err:
+        raise web.HTTPInternalServerError(reason=err.roll_up) from err
+    return web.json_response(result)
+
+
+async def register(app: web.Application):
+    """Register routes."""
+
+    app.add_routes(
+        [
+            web.get(
+                "/resolver/resolve/{did}",
+                resolve_did,
+                allow_head=False,
+            ),
+        ]
+    )
+
+
+def post_process_routes(app: web.Application):
+    """Amend swagger API."""
+
+    # Add top-level tags description
+    if "tags" not in app._state["swagger_dict"]:
+        app._state["swagger_dict"]["tags"] = []
+    app._state["swagger_dict"]["tags"].append(
+        {
+            "name": "resolver",
+            "description": "did resolver interface.",
+            "externalDocs": {"description": "Specification"},  # , "url": SPEC_URI},
+        }
+    )
diff --git a/aries_cloudagent/resolver/tests/__init__.py b/aries_cloudagent/resolver/tests/__init__.py
new file mode 100644
index 0000000000..8cc4eaaa56
--- /dev/null
+++ b/aries_cloudagent/resolver/tests/__init__.py
@@ -0,0 +1,43 @@
+DOC = {
+    "@context": "https://w3id.org/did/v1",
+    "id": "did:example:1234abcd",
+    "verificationMethod": [
+        {
+            "id": "did:example:1234abcd#4",
+            "type": "RsaVerificationKey2018",
+            "controller": "did:example:1234abcd",
+            "publicKeyPem": "-----BEGIN PUBLIC X…",
+        },
+        {
+            "id": "did:example:1234abcd#5",
+            "type": "RsaVerificationKey2018",
+            "controller": "did:example:1234abcd",
+            "publicKeyPem": "-----BEGIN PUBLIC 9…",
+        },
+        {
+            "id": "did:example:1234abcd#6",
+            "type": "RsaVerificationKey2018",
+            "controller": "did:example:1234abcd",
+            "publicKeyPem": "-----BEGIN PUBLIC A…",
+        },
+    ],
+    "authentication": [
+        {
+            "id": "did:example:123456789abcdefghi#ted",
+            "controller": "did:example:1234abcd",
+            "type": "RsaSignatureAuthentication2018",
+            "publicKey": "did:example:1234abcd#4",
+        },
+        "did:example:123456789abcdefghi#5",
+    ],
+    "service": [
+        {
+            "id": "did:example:123456789abcdefghi#did-communication",
+            "type": "did-communication",
+            "priority": 0,
+            "recipientKeys": ["did:example:1234abcd#4"],
+            "routingKeys": ["did:example:1234abcd#6"],
+            "serviceEndpoint": "did:example:xd45fr567794lrzti67;did-communication",
+        }
+    ],
+}
diff --git a/aries_cloudagent/resolver/tests/test_base.py b/aries_cloudagent/resolver/tests/test_base.py
new file mode 100644
index 0000000000..192eaa19d3
--- /dev/null
+++ b/aries_cloudagent/resolver/tests/test_base.py
@@ -0,0 +1,56 @@
+"""Test Base DID Resolver methods."""
+
+import pytest
+
+from pydid import DIDDocument
+
+from ..base import BaseDIDResolver, DIDMethodNotSupported, ResolverType
+
+
+class ExampleDIDResolver(BaseDIDResolver):
+    """Test DID Resolver."""
+
+    def __init__(self):
+        super().__init__()
+
+    async def setup(self, context):
+        pass
+
+    @property
+    def supported_methods(self):
+        return ["test"]
+
+    async def _resolve(self, profile, did) -> DIDDocument:
+        return DIDDocument("did:example:123")
+
+
+@pytest.fixture
+def native_resolver():
+    resolver = ExampleDIDResolver()
+    resolver.type = ResolverType.NATIVE
+    yield resolver
+
+
+@pytest.fixture
+def non_native_resolver():
+    yield ExampleDIDResolver()
+
+
+def test_native_on_native(native_resolver):
+    assert native_resolver.native is True
+
+
+def test_native_on_non_native(non_native_resolver):
+    assert non_native_resolver.native is False
+
+
+def test_supports(native_resolver):
+    assert native_resolver.supports("test") is True
+    assert native_resolver.supports("not supported") is False
+
+
+@pytest.mark.asyncio
+async def test_resolve_x(native_resolver):
+    with pytest.raises(DIDMethodNotSupported) as x_did:
+        await native_resolver.resolve(None, "did:nosuchmethod:xxx")
+    assert "does not support DID method" in str(x_did.value)
diff --git a/aries_cloudagent/resolver/tests/test_did_resolver.py b/aries_cloudagent/resolver/tests/test_did_resolver.py
new file mode 100644
index 0000000000..1f62f11037
--- /dev/null
+++ b/aries_cloudagent/resolver/tests/test_did_resolver.py
@@ -0,0 +1,177 @@
+"""Test did resolver registry."""
+
+import unittest
+
+import pytest
+from asynctest import mock as async_mock
+
+from ...resolver.base import (
+    BaseDIDResolver,
+    DIDMethodNotSupported,
+    DIDNotFound,
+    ResolverType,
+)
+from pydid import DID, DIDDocument, VerificationMethod
+from . import DOC
+from ..did_resolver import DIDResolver
+from ..did_resolver_registry import DIDResolverRegistry
+
+TEST_DID0 = "did:sov:Kkyqu7CJFuQSvBp468uaDe"
+TEST_DID1 = "did:btcr:8kyt-fzzq-qpqq-ljsc-5l"
+TEST_DID2 = "did:ethr:mainnet:0xb9c5714089478a327f09197987f16f9e5d936e8a"
+TEST_DID3 = "did:ion:EiDahaOGH-liLLdDtTxEAdc8i-cfCz-WUcQdRJheMVNn3A"
+TEST_DID4 = "did:github:ghdid"
+
+TEST_DIDS = [
+    TEST_DID0,
+    TEST_DID1,
+    TEST_DID2,
+    TEST_DID3,
+    TEST_DID4,
+]
+
+TEST_DID_METHOD0 = "sov"
+TEST_DID_METHOD1 = "btcr"
+TEST_DID_METHOD2 = "ethr"
+TEST_DID_METHOD3 = "ion"
+TEST_DID_METHOD4 = "github"
+
+TEST_DID_METHODS = [
+    TEST_DID_METHOD0,
+    TEST_DID_METHOD1,
+    TEST_DID_METHOD2,
+    TEST_DID_METHOD3,
+    TEST_DID_METHOD4,
+    "example",
+]
+
+TEST_METHOD_SPECIFIC_ID0 = "Kkyqu7CJFuQSvBp468uaDe"
+TEST_METHOD_SPECIFIC_ID1 = "8kyt-fzzq-qpqq-ljsc-5l"
+TEST_METHOD_SPECIFIC_ID2 = "mainnet:0xb9c5714089478a327f09197987f16f9e5d936e8a"
+TEST_METHOD_SPECIFIC_ID3 = "EiDahaOGH-liLLdDtTxEAdc8i-cfCz-WUcQdRJheMVNn3A"
+TEST_METHOD_SPECIFIC_ID4 = "ghdid"
+
+TEST_METHOD_SPECIFIC_IDS = [
+    TEST_METHOD_SPECIFIC_ID0,
+    TEST_METHOD_SPECIFIC_ID1,
+    TEST_METHOD_SPECIFIC_ID2,
+    TEST_METHOD_SPECIFIC_ID3,
+    TEST_METHOD_SPECIFIC_ID4,
+]
+
+
+class MockResolver(BaseDIDResolver):
+    def __init__(self, supported_methods, resolved=None, native: bool = False):
+        super().__init__(ResolverType.NATIVE if native else ResolverType.NON_NATIVE)
+        self._supported_methods = supported_methods
+        self.resolved = resolved
+
+    async def setup(self, context):
+        pass
+
+    @property
+    def supported_methods(self):
+        return self._supported_methods
+
+    async def _resolve(self, profile, did):
+        if isinstance(self.resolved, Exception):
+            raise self.resolved
+        return self.resolved.serialize()
+
+
+@pytest.fixture
+def resolver():
+    did_resolver_registry = DIDResolverRegistry()
+    for method in TEST_DID_METHODS:
+        resolver = MockResolver([method], DIDDocument.deserialize(DOC))
+        did_resolver_registry.register(resolver)
+    return DIDResolver(did_resolver_registry)
+
+
+@pytest.fixture
+def profile():
+    yield async_mock.MagicMock()
+
+
+def test_create_resolver(resolver):
+    assert len(resolver.did_resolver_registry.resolvers) == len(TEST_DID_METHODS)
+
+
+@pytest.mark.parametrize("did, method", zip(TEST_DIDS, TEST_DID_METHODS))
+def test_match_did_to_resolver(resolver, did, method):
+    did = DID(did)
+    base_resolver, *_ = resolver._match_did_to_resolver(did)
+    assert base_resolver.supports(method)
+
+
+def test_match_did_to_resolver_x_not_supported(resolver):
+    did = DID("did:cowsay:EiDahaOGH-liLLdDtTxEAdc8i-cfCz-WUcQdRJheMVNn3A")
+    with pytest.raises(DIDMethodNotSupported):
+        resolver._match_did_to_resolver(did)
+
+
+def test_match_did_to_resolver_native_priority():
+    registry = DIDResolverRegistry()
+    native = MockResolver(["sov"], native=True)
+    non_native = MockResolver(["sov"], native=False)
+    registry.register(non_native)
+    registry.register(native)
+    resolver = DIDResolver(registry)
+    assert [native, non_native] == resolver._match_did_to_resolver(DID(TEST_DID0))
+
+
+def test_match_did_to_resolver_registration_order():
+    registry = DIDResolverRegistry()
+    native1 = MockResolver(["sov"], native=True)
+    registry.register(native1)
+    native2 = MockResolver(["sov"], native=True)
+    registry.register(native2)
+    non_native3 = MockResolver(["sov"], native=False)
+    registry.register(non_native3)
+    native4 = MockResolver(["sov"], native=True)
+    registry.register(native4)
+    resolver = DIDResolver(registry)
+    assert [native1, native2, native4, non_native3] == resolver._match_did_to_resolver(
+        DID(TEST_DID0)
+    )
+
+
+@pytest.mark.asyncio
+async def test_dereference(resolver, profile):
+    url = "did:example:1234abcd#4"
+    expected: dict = DOC["verificationMethod"][0]
+    actual: VerificationMethod = await resolver.dereference(profile, url)
+    assert expected == actual.serialize()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("did", TEST_DIDS)
+async def test_resolve(resolver, profile, did):
+    did_doc = await resolver.resolve(profile, did)
+    assert isinstance(did_doc, DIDDocument)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("did", TEST_DIDS)
+async def test_resolve_did(resolver, profile, did):
+    did = DID(did)
+    did_doc = await resolver.resolve(profile, did)
+    assert isinstance(did_doc, DIDDocument)
+
+
+@pytest.mark.asyncio
+async def test_resolve_did_x_not_supported(resolver, profile):
+    did = DID("did:cowsay:EiDahaOGH-liLLdDtTxEAdc8i-cfCz-WUcQdRJheMVNn3A")
+    with pytest.raises(DIDMethodNotSupported):
+        await resolver.resolve(profile, did)
+
+
+@pytest.mark.asyncio
+async def test_resolve_did_x_not_found(profile):
+    did = DID("did:cowsay:EiDahaOGH-liLLdDtTxEAdc8i-cfCz-WUcQdRJheMVNn3A")
+    cowsay_resolver_not_found = MockResolver("cowsay", resolved=DIDNotFound())
+    registry = DIDResolverRegistry()
+    registry.register(cowsay_resolver_not_found)
+    resolver = DIDResolver(registry)
+    with pytest.raises(DIDNotFound):
+        await resolver.resolve(profile, did)
diff --git a/aries_cloudagent/resolver/tests/test_did_resolver_registry.py b/aries_cloudagent/resolver/tests/test_did_resolver_registry.py
new file mode 100644
index 0000000000..dba7afffbd
--- /dev/null
+++ b/aries_cloudagent/resolver/tests/test_did_resolver_registry.py
@@ -0,0 +1,12 @@
+"""Test did resolver registery."""
+
+import pytest
+import unittest
+from ..did_resolver_registry import DIDResolverRegistry
+
+
+def test_create_registry():
+    did_resolver_registry = DIDResolverRegistry()
+    test_resolver = unittest.mock.MagicMock()
+    did_resolver_registry.register(test_resolver)
+    assert did_resolver_registry.resolvers == [test_resolver]
diff --git a/aries_cloudagent/resolver/tests/test_routes.py b/aries_cloudagent/resolver/tests/test_routes.py
new file mode 100644
index 0000000000..de8c1dad5e
--- /dev/null
+++ b/aries_cloudagent/resolver/tests/test_routes.py
@@ -0,0 +1,93 @@
+"""Test resolver routes."""
+
+# pylint: disable=redefined-outer-name
+
+import pytest
+from asynctest import mock as async_mock
+from pydid import DIDDocument
+
+from ...admin.request_context import AdminRequestContext
+from .. import routes as test_module
+from ..base import (
+    DIDMethodNotSupported,
+    DIDNotFound,
+    ResolverError,
+)
+from ..did_resolver import DIDResolver
+from . import DOC
+
+did_doc = DIDDocument.deserialize(DOC)
+
+
+@pytest.fixture
+def mock_response():
+    json_response = async_mock.MagicMock()
+    temp_value = test_module.web.json_response
+    test_module.web.json_response = json_response
+    yield json_response
+    test_module.web.json_response = temp_value
+
+
+@pytest.fixture
+def mock_resolver():
+    did_resolver = async_mock.MagicMock()
+    did_resolver.resolve = async_mock.CoroutineMock(return_value=did_doc)
+    yield did_resolver
+
+
+@pytest.fixture
+def mock_request(mock_resolver):
+    context = AdminRequestContext.test_context({DIDResolver: mock_resolver})
+    outbound_message_router = async_mock.CoroutineMock()
+    request_dict = {
+        "context": context,
+        "outbound_message_router": outbound_message_router,
+    }
+    request = async_mock.MagicMock(
+        match_info={
+            "did": "did:ethr:mainnet:0xb9c5714089478a327f09197987f16f9e5d936e8a",
+        },
+        query={},
+        json=async_mock.CoroutineMock(return_value={}),
+        __getitem__=lambda _, k: request_dict[k],
+    )
+    yield request
+
+
+@pytest.mark.asyncio
+async def test_resolver(mock_request, mock_response):
+    await test_module.resolve_did(mock_request)
+    mock_response.assert_called_once_with(did_doc.serialize())
+    # TODO: test http response codes
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "side_effect, error",
+    [
+        (DIDNotFound, test_module.web.HTTPNotFound),
+        (DIDMethodNotSupported, test_module.web.HTTPNotImplemented),
+        (ResolverError, test_module.web.HTTPInternalServerError),
+    ],
+)
+async def test_resolver_not_found_error(
+    mock_resolver, mock_request, side_effect, error
+):
+    mock_resolver.resolve = async_mock.CoroutineMock(side_effect=side_effect())
+    with pytest.raises(error):
+        await test_module.resolve_did(mock_request)
+
+
+@pytest.mark.asyncio
+async def test_register():
+    mock_app = async_mock.MagicMock()
+    mock_app.add_routes = async_mock.MagicMock()
+    await test_module.register(mock_app)
+    mock_app.add_routes.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_post_process_routes():
+    mock_app = async_mock.MagicMock(_state={"swagger_dict": {}})
+    test_module.post_process_routes(mock_app)
+    assert "tags" in mock_app._state["swagger_dict"]
diff --git a/requirements.txt b/requirements.txt
index 5ff2b37a62..9fb55063cc 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -16,3 +16,4 @@ py_multicodec==0.2.1
 pyyaml~=5.3.1
 ConfigArgParse~=1.2.3
 pyjwt~=1.7.1
+pydid==0.2.3