This guide will walk you through installing and using aas-http-client .
Prerequisites:
- Python 3.10 or newer
- Access to an AAS server endpoint
Install via pip:
pip install aas-http-clientFor detailed configuration options, authentication methods and examples, see the Configuration Guide.
This chapter provides a practical, end-to-end overview of working with aas-http-client after installation.
It starts with client and wrapper creation patterns, then walks through the most relevant repository and registry endpoint groups, including experimental attachment endpoints.
Use the sections below as quick-start references and adapt the examples to your server setup, authentication mode, and ID encoding configuration.
For a complete list and detailed descriptions of all functions, see the API reference.
Most important notes before calling endpoint methods:
- Logical groups map to attributes on the client instance:
client.shells-> Shell repository endpointsclient.submodels-> Submodel repository endpointsclient.shell_registry-> Shell descriptor registry endpointsclient.submodel_registry-> Submodel descriptor registry endpointsclient.experimental-> Experimental attachment endpoints
- For the wrapper, most endpoint implementations are available directly on
wrapper(SDK object responses). Registry endpoints are currently not exposed onwrapper. encoded_idsbehavior:- Keep
encoded_idsaligned with your server expectations. - If
encoded_idsis enabled (default), pass encoded IDs for endpoint path parameters when required by the API/server. - Example (encoded mode):
submodel_identifier="dXJuOmV4YW1wbGU6c3VibW9kZWw6MDAx". - If
encoded_idsis disabled, the client encodes IDs internally for endpoints that require encoded path segments. - Example (decoded mode):
submodel_identifier="urn:example:submodel:001". - Inconsistent
encoded_idssettings are a common reason for404responses.
- Keep
- Pagination:
- List endpoints commonly support
limitandcursor. - Start with a sensible
limit(for example:10or100), then continue with the returned cursor until no more results are available. - For client responses, read elements from
result(for example:response.get("result", [])).
- List endpoints commonly support
- Return value conventions:
- Read operations return
dict, SDK objects, orNoneon failure/not found. - Write/delete operations usually return
bool(Trueon success,Falseon failure). - Always handle
NoneandFalseexplicitly in production code.
- Read operations return
- Wrapper vs client:
- Use
client.*for raw dictionary responses and full registry coverage. - Use
wrapper.*for SDK object-oriented handling. - Registry endpoints are currently not exposed on wrapper (use
client.shell_registryandclient.submodel_registry).
- Use
- Experimental endpoints:
- Attachment endpoints are experimental and may vary by server implementation.
- Prefer feature checks or graceful fallbacks when targeting multiple server products/versions.
There are three ways to create an AAS HTTP client or wrapper.
Use either a client (dictionary-based API) or a wrapper (SDK object-based API), depending on your use case.
Creation methods can return None if the configuration is invalid or the connection fails.
For production usage, avoid hardcoding secrets in source code. Load credentials from environment variables or a secret manager.
Create a client or wrapper by providing parameters directly using create_by_url :
import os
from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper
client = aas_client.create_by_url(
base_url="http://localhost:8080",
basic_auth_username="admin",
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", ""),
time_out=300,
ssl_verify=True
)
wrapper = sdk_wrapper.create_by_url(
base_url="http://localhost:8080",
basic_auth_username="admin",
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", ""),
time_out=300,
ssl_verify=True
)
if client is None:
raise RuntimeError("Client creation failed")
if wrapper is None:
raise RuntimeError("Wrapper creation failed")
print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)Create a client or wrapper using a configuration dictionary with create_by_dict :
import os
from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper
config = {
"BaseUrl": "http://localhost:8080",
"TimeOut": 300,
"AuthenticationSettings": {
"BasicAuth": {
"Username": "admin"
}
}
}
client = aas_client.create_by_dict(
configuration=config,
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)
wrapper = sdk_wrapper.create_by_dict(
configuration=config,
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)
if client is None:
raise RuntimeError("Client creation failed")
if wrapper is None:
raise RuntimeError("Wrapper creation failed")
print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)Create a client or wrapper using a JSON configuration file with create_by_config :
import os
from pathlib import Path
from aas_http_client.classes.client import aas_client
from aas_http_client.classes.wrapper import sdk_wrapper
config_file = Path("config.json")
client = aas_client.create_by_config(
config_file=config_file,
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)
wrapper = sdk_wrapper.create_by_config(
config_file=config_file,
basic_auth_password=os.getenv("AAS_BASIC_AUTH_PASSWORD", "")
)
if client is None:
raise RuntimeError("Client creation failed")
if wrapper is None:
raise RuntimeError("Wrapper creation failed")
print("Client connectivity:", client.get_root() is not None)
print("Wrapper connectivity:", wrapper.get_client().get_root() is not None)This section shows how to work with the most common Shell repository operations after client or wrapper creation.
Most important points:
- Shell endpoints are available via
client.shells(dictionary responses) or directly onwrapper(SDK object responses). - List endpoints are paginated. Use
limitandcursorwhen iterating through larger result sets. - Always handle
Noneresults to detect connectivity, authorization, or server-side issues.
For the full list of available methods and signatures, see:
# Assumes `client` was created successfully in one of the sections above.
result = client.shells.get_all_asset_administration_shells(limit=10)
if result is None:
raise RuntimeError("Failed to fetch shells")
shells = result.get("result", [])
print(f"Received {len(shells)} shell(s)")# Assumes `wrapper` was created successfully in one of the sections above.
aas_id = "urn:example:aas:001"
aas = wrapper.get_asset_administration_shell_by_id(aas_id)
if aas is None:
print("Shell not found or request failed")
else:
print("Found shell with id:", aas.id)This section shows how to work with common Submodel repository operations after client or wrapper creation.
Most important points:
- Submodel endpoints are available via
client.submodels(dictionary responses) or directly onwrapper(SDK object responses). - List endpoints are paginated. Use
limitandcursorwhen retrieving larger result sets. levelandextentcan be used to control response depth and blob behavior.- Always handle
Noneresults to detect connectivity, authorization, or server-side issues.
For the full list of available methods and signatures, see:
# Assumes `client` was created successfully in one of the sections above.
result = client.submodels.get_all_submodels(limit=10)
if result is None:
raise RuntimeError("Failed to fetch submodels")
submodels = result.get("result", [])
print(f"Received {len(submodels)} submodel(s)")# Assumes `wrapper` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
submodel = wrapper.get_submodel_by_id(submodel_id)
if submodel is None:
print("Submodel not found or request failed")
else:
print("Found submodel with id:", submodel.id)This section shows how to work with Asset Administration Shell registry endpoints.
Most important points:
- Shell descriptor endpoints are available via
client.shell_registry. - Registry endpoints currently are not exposed on
wrapper. - List endpoints are paginated. Use
limitandcursorwhen iterating through larger result sets. - Depending on server settings, IDs may need Base64 URL-safe encoding. Keep
encoded_idsconsistent with your server setup.
For the full list of available methods and signatures, see:
# Assumes `client` was created successfully in one of the sections above.
result = client.shell_registry.get_all_asset_administration_shell_descriptors(limit=10)
if result is None:
raise RuntimeError("Failed to fetch shell descriptors")
descriptors = result.get("result", [])
print(f"Received {len(descriptors)} shell descriptor(s)")# Assumes `client` was created successfully in one of the sections above.
aas_id = "urn:example:aas:001"
descriptor = client.shell_registry.get_asset_administration_shell_descriptor_by_id(aas_id)
if descriptor is None:
print("Shell descriptor not found or request failed")
else:
print("Found descriptor with id:", descriptor.get("id"))# Assumes `client` was created successfully in one of the sections above.
payload = {
"id": "urn:example:aas:001",
"idShort": "aas_example_001",
"assetKind": "Instance",
"endpoints": [
{
"interface": "AAS-3.0",
"protocolInformation": {
"href": "http://localhost:8080/shells/urn:example:aas:001",
"endpointProtocol": "http",
"endpointProtocolVersion": ["1.1"],
"subprotocol": "",
"subprotocolBody": "",
"subprotocolBodyEncoding": "plain",
"securityAttributes": []
}
}
],
"submodelDescriptors": []
}
created = client.shell_registry.post_asset_administration_shell_descriptor(payload)
if created is None:
raise RuntimeError("Failed to create shell descriptor")
print("Registered descriptor:", created.get("id"))This section shows how to work with Submodel registry endpoints.
Most important points:
- Submodel descriptor endpoints are available via
client.submodel_registry. - Registry endpoints currently are not exposed on
wrapper. - List endpoints are paginated. Use
limitandcursorwhen iterating through larger result sets.
For the full list of available methods and signatures, see:
# Assumes `client` was created successfully in one of the sections above.
result = client.submodel_registry.get_all_submodel_descriptors(limit=10)
if result is None:
raise RuntimeError("Failed to fetch submodel descriptors")
descriptors = result.get("result", [])
print(f"Received {len(descriptors)} submodel descriptor(s)")# Assumes `client` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
descriptor = client.submodel_registry.get_submodel_descriptor_by_id(submodel_id)
if descriptor is None:
print("Submodel descriptor not found or request failed")
else:
print("Found descriptor with id:", descriptor.get("id"))# Assumes `client` was created successfully in one of the sections above.
payload = {
"id": "urn:example:submodel:001",
"idShort": "sm_example_001",
"endpoints": [
{
"interface": "SUBMODEL-3.0",
"protocolInformation": {
"href": "http://localhost:8080/submodels/urn:example:submodel:001",
"endpointProtocol": "http",
"endpointProtocolVersion": ["1.1"],
"subprotocol": "",
"subprotocolBody": "",
"subprotocolBodyEncoding": "plain",
"securityAttributes": []
}
}
],
"semanticId": {
"type": "ExternalReference",
"keys": [
{
"type": "GlobalReference",
"value": "urn:example:semantic:submodel:001"
}
]
}
}
created = client.submodel_registry.post_submodel_descriptor(payload)
if created is None:
raise RuntimeError("Failed to create submodel descriptor")
print("Registered descriptor:", created.get("id"))This section shows the experimental attachment endpoints for file content on Submodel File elements.
Most important points:
- Experimental endpoints are available via
client.experimentaland via wrapper methods prefixed withexperimental_. - Experimental endpoints may not be supported by every server implementation.
For the full list of available methods and signatures, see:
from pathlib import Path
# Assumes `wrapper` was created successfully in one of the sections above.
attachment = wrapper.experimental_get_file_by_path_submodel_repo(
submodel_identifier="urn:example:submodel:001",
id_short_path="Documents.UserManual"
)
if attachment is None:
print("No attachment found or request failed")
else:
output = Path("downloaded_" + attachment.filename)
output.write_bytes(attachment.content)
print("Downloaded attachment to:", output)
print("Detected content type:", attachment.content_type)from pathlib import Path
# Assumes `client` was created successfully in one of the sections above.
submodel_id = "urn:example:submodel:001"
id_short_path = "Documents.UserManual"
file_path = Path("manual.pdf")
created = client.experimental.post_file_by_path_submodel_repo(
submodel_identifier=submodel_id,
id_short_path=id_short_path,
file=file_path
)
if not created:
raise RuntimeError("Failed to upload attachment with POST")
updated = client.experimental.put_file_by_path_submodel_repo(
submodel_identifier=submodel_id,
id_short_path=id_short_path,
file=file_path
)
if not updated:
raise RuntimeError("Failed to replace attachment with PUT")
print("Attachment upload/update successful")This section covers the low-level generic endpoint helpers on AasHttpClient .
Most important points:
- Generic endpoint helpers are available on
clientonly (not onwrapper). - They allow direct HTTP calls to arbitrary endpoint URLs when no specialized helper exists.
# Assumes `client` was created successfully in one of the sections above.
url = f"{client.base_url}/description"
result = client.get_endpoint(url)
if result is None:
print("GET request failed or returned non-200")
else:
print("Service description keys:", list(result.keys()))# Assumes `client` was created successfully in one of the sections above.
url = f"{client.base_url}/some-custom-endpoint"
payload = {"example": "value"}
post_error = client.post_endpoint(url, payload)
if post_error is not None:
print("POST returned non-success:", post_error)
put_error = client.put_endpoint(url, payload)
if put_error is not None:
print("PUT returned non-success:", put_error)
patch_error = client.patch_endpoint(url, {"example": "updated"})
if patch_error is not None:
print("PATCH returned non-success:", patch_error)
delete_error = client.delete_endpoint(url)
if delete_error is not None:
print("DELETE returned non-success:", delete_error)For the full list of available methods and signatures, see: