tractusx-sdk 0.8.1rc1

Eclipse Tractus-X Software Development KIT - The Dataspace & Industry Foundation Middleware

Tractus-X Software Development KIT (SDK)

Eclipse Tractus-X Software Development KIT - The Dataspace & Industry Foundation Libraries

A modular library/toolbox/framework, that helps you to build applications/scripts/microservices which can use the Connector, DTR, Discovery Services for providing and consuming data (any data).

It aims to provide a reference implementation for the various interactions between applications and services like the EDC, Digital Twin Registry and Submodel Service. This "tool box" helps you to provide data and consume data on a Dataspace Context, how you orchestrate it is then up to you and your use case.

Additionally to this various microservices are provided at the eclipse-tractusx/tractusx-sdk-services repository, giving an lighthouse implementation on how to use the SDK for various purposes. Other examples of solutions which are built with this SDK is the Industry Core Hub.

This SDK will manage automatically the version updates from the EDC and the Digital Twin Registry, that will be maintained by the community.

No specific use case logic will be configured here, only the bare minimum for interacting in a Dataspace and developing your own applications with this stack, based on the KITs which adopt the core data exchange functionalities, in concrete the following ones:

• Connector KIT
• Digital Twin KIT
• Industry Core KIT

Installation

Install the package directly from PyPI https://pypi.org/p/tractusx_sdk:

pip install tractusx-sdk

Documentation

Our latest documentation can be found here:

https://eclipse-tractusx.github.io/tractusx-sdk/main/

Usage

You can find some examples in the dedicated section.

[!NOTE] We are working to make the SDK documentation more accessible and easy to adopt. Our objective is to have an "Open Specification" at our github pages domain. Support us by joining our open meetings or contributing with documentation.

We have an advanced usage guide in docs/user.

Here we also have a quick guide:

📖 Quick Guide:

• Consumption - Consuming data from the dataspace
  • Call endpoints with one click - Quick data access
  • Get access to endpoints behind connectors & reuse connections - Easy connection management
  • Want more? - Advanced catalog operations
  • Advanced Usage - Low-level contract negotiations
  • Discovery - Finding connectors and services
• Provision - Setting up data provision and asset creation
  • Digital Twin Registry - Working with AAS shell descriptors
• All in one - Combined provider and consumer usage

Consumption

from tractusx_sdk.dataspace.managers.connection import PostgresMemoryRefreshConnectionManager, FileSystemConnectionManager, MemoryConnectionManager
from tractusx_sdk.dataspace.services.connector import ServiceFactory
from tractusx_sdk.dataspace.services import BaseConnectorService

#####################################################

# Select a connection manager

# Postgres Synced Option with Memory
from sqlmodel import create_engine

engine = create_engine(str("postgresql://user:password@localhost:5432/db"))
connection_manager = PostgresMemoryRefreshConnectionManager(
    engine=engine, logger=logger, verbose=True)

# Another Postgres-Only option is available but is not recommended

# FileSystem
connection_manager = FileSystemConnectionManager(
    path="./data/my-connections.json", logger=logger, verbose=True)

# Memory Only
connection_manager = MemoryConnectionManager(logger=logger, verbose=True)

#####################################################

consumer_connector_headers: dict = {
    "X-Api-Key": "my-api-key",
    "Content-Type": "application/json"
}

consumer_connector_service: BaseConnectorService = ServiceFactory.get_connector_consumer_service(
    dataspace_version="jupiter",  # "saturn" is also
    base_url="https://my-connector-controlplane.url",
    dma_path="/management",
    headers=consumer_connector_headers,
    connection_manager=connection_manager,  # Select one from above
    logger=logger,  # You can remove this to keep logs disabled
    verbose=True  # You can remove this to keep logs disabled
)

Call endpoints with one click

policies_to_accept: list = [{"odrl:permission": {"odrl:action": {"@id": "odrl:use"}, "odrl:constraint": {"odrl:and": [{"odrl:leftOperand": {"@id": "cx-policy:FrameworkAgreement"}, "odrl:operator": {"@id": "odrl:eq"}, "odrl:rightOperand": "DataExchangeGovernance:1.0"}, {"odrl:leftOperand": {
    "@id": "cx-policy:Membership"}, "odrl:operator": {"@id": "odrl:eq"}, "odrl:rightOperand": "active"}, {"odrl:leftOperand": {"@id": "cx-policy:UsagePurpose"}, "odrl:operator": {"@id": "odrl:eq"}, "odrl:rightOperand": "cx.core.digitalTwinRegistry:1"}]}}, "odrl:prohibition": [], "odrl:obligation": []}]

digital_twins = consumer_connector_service.do_get(
    counter_party_id="BPNL00000003AYRE",
    counter_party_address="https://connector-edc-controlplane.tractusx.io/api/v1/dsp",
    filter_expression=consumer_connector_service.get_filter_expression(
        key="'http://purl.org/dc/terms/type'.'@id'",
        operator="=",
        value="https://w3id.org/catenax/taxonomy#DigitalTwinRegistry"
    ),
    path="/shell-descriptors",
    params={"limit": 5},
    policies=policies_to_accept
)

# Here are the digital twins
print(digital_twins.json())

Get access to the endpoints behind the connector & reuse connections

You can do more! Explore other methods from the service.

# Don't want to make the one click approach? Feel free to implement your own logic

dataplane_url, access_token = consumer_connector_service.do_dsp(
    counter_party_id="BPNL00000003AYRE",
    counter_party_address="https://connector-edc-controlplane.tractusx.io/api/v1/dsp",
    filter_expression=consumer_connector_service.get_filter_expression(
        key="'http://purl.org/dc/terms/type'.'@id'",
        operator="=",
        value="https://w3id.org/catenax/taxonomy#DigitalTwinRegistry"
    ),
    policies=policies_to_accept
)

# Using the dataplane proxy and the access token, you can call how much times you like different APIs behind the proxy, reusing the open connection!

Want more?

# Get Catalog by yourself

catalog = consumer_connector_service.get_catalog_by_dct_type(
    dct_type="https://w3id.org/catenax/taxonomy#DigitalTwinRegistry",
    counter_party_id="BPNL00000003AYRE",
    counter_party_address="https://edc1-controlplane/api/v1/dsp",
    timeout=15
)

# One is not enough? Call in parallel!
catalogs = consumer_connector_service.get_catalogs_by_dct_type(
    dct_type="https://w3id.org/catenax/taxonomy#DigitalTwinRegistry",
    counter_party_id="BPNL00000003AYRE",
    edcs=[
        "https://edc1-controlplane/api/v1/dsp",
        "https://edc2-controlplane/api/v1/dsp",
        "https://edc3-controlplane/api/v1/dsp"
    ],
    timeout=15
)

Advanced Usage

# You can even implement it in the granularity you want!
from tractusx_sdk.dataspace.models.connector.base_contract_negotiation_model import BaseContractNegotiationModel
from tractusx_sdk.dataspace.models.connector.model_factory import ModelFactory

request: BaseContractNegotiationModel = ModelFactory. get_contract_negotiation_model(
    dataspace_version="jupiter",
    context=[
        "https://w3id.org/tractusx/policy/v1.0.0",
        "http://www.w3.org/ns/odrl.jsonld",
        {
            "@vocab": "https://w3id.org/edc/v0.0.1/ns/"
        }
    ],
    counter_party_address="https://edc1-controlplane/api/v1/dsp",
    offer_id="offer-id",
    asset_id="asset-id",
    provider_id="BPNL00000003AYRE",
    # Add here the policy you want to accept
    offer_policy=policies_to_accept[0]
)

# Build catalog api url
response: Response = self.edrs.create(request)
# In case the response code is not successful or the response is null
if (response is None or response.status_code != 200):
    return None

negotiation_id = response.json().get("@id", None)

response: Response = self.edrs.get_data_address(
    oid=negotiation_id, params={"auto_refresh": True})

Discovery

from tractusx_sdk.dataspace.managers import OAuth2Manager
from tractusx_sdk.dataspace.services.discovery import ConnectorDiscoveryService, DiscoveryFinderService

discovery_oauth = OAuth2Manager(
    auth_url="https://central-idp/auth/",
    realm="CX-Central",
    clientid="456as2id",
    clientsecret="asbadjsk2as4sad574s",
)

discovery_finder_service = DiscoveryFinderService(
    url="https://discovery-finder.url/api/v1.0/administration/connectors/discovery/search",
    oauth=discovery_oauth
)

# Create the connector discovery service for the consumer
connector_discovery_service = ConnectorDiscoveryService(
    oauth=discovery_oauth,
    discovery_finder_service=discovery_finder_service
)

connector_discovery_service.find_connector_by_bpn("BPNL00000000TS1D")

Provision

from tractusx_sdk.dataspace.services.connector import ServiceFactory, BaseConnectorService

provider_connector_headers: dict = {
    "X-Api-Key": "my-api-key",
    "Content-Type": "application/json"
}

# Create the connector provider service
provider_connector_service: BaseConnectorService = ServiceFactory.get_connector_provider_service(
    dataspace_version="jupiter",  # "saturn" is also available
    base_url="https://my-connector-controlplane.url",
    dma_path="/management",
    headers=provider_connector_headers,
    logger=logger,  # You can remove this to keep logs disabled
    verbose=True  # You can remove this to keep logs disabled
)
provider_connector_service.create_asset(
    asset_id="my-asset-id",
    base_url="https://submodel-service.url/",
    dct_type="cx-taxo:SubmodelBundle",
    version="3.0",
    semantic_id="urn:samm:io.catenax.part_type_information:1.0.0#PartTypeInformation",
    headers={
        "X-Api-Key": "my-secret-to-the-submodel-service"
    }
)
provider_connector_service.create_asset(
    asset_id="digital-twin-registry",
    base_url="https://digital-twin-registry.tractusx.io/api/v3",
    dct_type="https://w3id.org/catenax/taxonomy#DigitalTwinRegistry",
    version="3.0",
    proxy_params={
        "proxyQueryParams": "true",
        "proxyPath": "true",
        "proxyMethod": "true",
        "proxyBody": "true"
    }
)

# And you can do much more! Create Policies, Contracts, etc

Digital Twin Registry

from tractusx_sdk.industry.models.aas.v3 import (
    ShellDescriptor, MultiLanguage, SpecificAssetId, AssetKind, ReferenceTypes, Reference, ReferenceKey, ReferenceKeyTypes
)
from tractusx_sdk.industry.services import AasService
aas_service = AasService(
    base_url="https://digital-twin-registry.tractusx.io",
    # Here you can add another one if you want.
    base_lookup_url="https://aas-discovery.tractusx.io",
    api_path="/api/v3",
)

# Get shells or one shell
existing_shell: ShellDescriptor = aas_service.get_asset_administration_shell_descriptor_by_id(
    aas_identifier="urn:uuid:ad7bc88c-fa31-40d8-8f17-2ceaf1295ff6"
)

########
# Create Shells

# Create display name with multiple languages
display_name_en = MultiLanguage(language="en", text="Vehicle Battery Shell")
display_name_de = MultiLanguage(language="de", text="Fahrzeugbatterie Shell")
display_names = [display_name_en, display_name_de]

# Create description with multiple languages
description_en = MultiLanguage(
    language="en", text="Digital twin shell for electric vehicle battery component")
description_de = MultiLanguage(
    language="de", text="Digitaler Zwilling für Elektrofahrzeug-Batteriekomponente")
descriptions = [description_en, description_de]

bpns_list = ["BPNL00000003AYRE", "BPNL000000000TEA4"]
manufacturer_id = "BPNL000000032ASTT"
# Create specific asset identifiers to allow shell descriptors to be seen.
manufacturer_part_id = SpecificAssetId(
    name="manufacturerPartId",
    value="BAT-12345-ABC",
    externalSubjectId=Reference(
        type=ReferenceTypes.EXTERNAL_REFERENCE,
        keys=[ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE, value=bpn) for bpn in bpns_list] or
        [ReferenceKey(type=ReferenceKeyTypes.GLOBAL_REFERENCE,
                      value=manufacturer_id)]
    )
)

vehicle_identification_number = SpecificAssetId(
    name="van",
    value="WVWZZZ1JZ3W386752"
)

specific_asset_ids = [manufacturer_part_id, vehicle_identification_number]

# Create the shell descriptor
shell = ShellDescriptor(
    id="urn:uuid:ad7bc88c-fa31-40d8-8f17-2ceaf1295ff6",
    idShort="VehicleBatteryShell001",
    displayName=display_names,
    description=descriptions,
    assetType="Battery",
    assetKind=AssetKind.INSTANCE,
    globalAssetId="urn:uuid:550e8400-e29b-41d4-a716-446655440000",
    specificAssetIds=specific_asset_ids,
)
res = aas_service.create_asset_administration_shell_descriptor(
    shell_descriptor=shell)

# Congrats! You created a shell according to the AAS 3.0 standards!

All in one

Combine provider and consumer in one module to take the best from your connector!

from tractusx_sdk.dataspace.services.connector import ServiceFactory

connector_service:dict = {
    "X-Api-Key": "my-api-key",
    "Content-Type": "application/json"
}
connector_service = ServiceFactory.get_connector_service(
      dataspace_version="jupiter", ## "saturn" is also available
      base_url="https://my-connector-controlplane.url",
      dma_path="/management",
      headers=connector_connector_headers,
      connection_manager=connection_manager,
      logger=logger, ## You can remove this to keep logs disabled
      verbose=True ## You can remove this to keep logs disabled
)

connector_service.consumer.get_catalog(...)
connector_service.provider.assets.create(...)

Test Compliance Kits (TCKs)

The tck/connector/ directory contains Test Compliance Kits for validating end-to-end connector functionality across different Eclipse Tractus-X EDC versions and deployment scenarios.

Available TCKs

Saturn Protocol (EDC v0.11.x - DSP 2025-1):

• tck_e2e_saturn_0-11-X_simple_did.py - Simple one-call using do_get() (DID-based)
• tck_e2e_saturn_0-11-X_detailed_did.py - Detailed step-by-step flow (DID-based)
• tck_e2e_umbrella_saturn_simple_did.py - Umbrella Helm chart (simple)
• tck_e2e_umbrella_saturn_detailed.py - Umbrella Helm chart (detailed)

Jupiter Protocol (EDC v0.8.x - 0.10.x - Legacy DSP):

• tck_e2e_jupiter_0-10-X_simple.py - Simple one-call (BPNL-based)
• tck_e2e_jupiter_0-10-X_detailed.py - Detailed step-by-step flow (BPNL-based)

Running TCKs

# Install dependencies
poetry install

# Run a TCK example
python tck/connector/tck_e2e_umbrella_saturn_simple_did.py

TCKs demonstrate the complete data exchange flow:

1. Provider: Provisions data (Asset + Policies + Contract Definition)
2. Consumer: Discovers, negotiates, transfers, and accesses data
3. Backend: Simple data storage for testing

Each TCK generates timestamped logs in logs/<tck_name>/<date>/ with request/response details, policy configurations, and test results.

Deployment Scenarios

Tractus-X Umbrella: TCKs are pre-configured with Tractus-X Umbrella Helm chart values, providing a complete Catena-X dataspace environment with EDC connectors, SSI DIM Wallet, Digital Twin Registry, and test data.

Standalone EDC: TCKs can be adapted for custom EDC deployments by updating connector URLs, API keys, and BPN/DID values.

Roadmap

The development roadmap is aligned with the Industry Core Hub. For the latest milestones and planned features, see the GitHub project board and open discussions.

[!NOTE] For known compatibility issues with specific connector versions, check the issues page.

What can you do with this SDK?

• You can create a frontend for your use case (with any technology) and build your own Backend Apis with this tool box.
• Furthermore, you can use this in a Jupyter notebook for example, or create your personal scripts for using the Connector, DTR and Submodel Service.
• It enables you to build your own use case logic over it, without needing to worry about versioning (Jupiter and Saturn supported).
• Base yourself in the Industry Core Hub which provides you a "lighthouse" for using this SDK.

Applications that use this SDK

• Industry Core Hub: Example on how to use the SDK to develop an application.

  • Backend: An example of how to use the SDK dataspace & industry libaries in your application.
  • Frontend: An example of how to use the SDK API interfaces from each microservice.
  • The use case add-ons from the IC-Hub will also use this!

• Tractus-X SDK Services: Repository where the reusable APIs/Services for the SDK are available as microservices.

  • DT Pull Service: Provides a service that pulls digital twins from data providers.
  • Test Orchestrator: Provides a test agent service that can check if the configuration of your data provision services is compliant with the standard schemas & syntaxis.

• Tractus-X AAS Suite: A integration project between the Tractus-X SDK (being used as client for the Connector) and the BaSyx Python SDK.

• Open for more collaboration!

Module Logic Architecture

To ease the understanding what is the tool box (SDK) here is a resumed diagram:

Why was this Tractus-X SDK Created?

Here you will find a design decision which was taken at the beginning of the industry core hub development:

Industry Core Hub Decision Record 0002

While developing the Industry Core Hub, in parallel we decided to create a SDK for Tractus-X.

Industry Core Hub Decision Record 0003 Create new Repository

Having an individual SDK repository, we are creating a reusable and modular middleware/library for all use cases and applications that want to easily interact with the Tractus-X Datapaces Components required for data provision and data consumption:

• Tractus-X Eclipse Dataspace Connector (EDC)
• Tractus-X Digital Twin Registry (DTR)
• Simple Data Backend

And other core services like:

Additional Services:

• Discovery Services:

  • Discovery Finder
  • BPN Discovery
  • EDC Discovery

• Portal IAM/IDP

Our aim is to automate the target releases and compatibility with this systems using DevOps mechanisms.

High Level Architecture

Providing reusable modules:

• Dataspace Foundation Library

  • Enables your "bytes" data exchange using the EDC and the core services from Catena-X.
  • It provides tools for your exchange agnostic from your use case.

• Industry Foundation Library

  • Enables your data exchange using the dataspace foundation library but for the usage of Digital Twins in the Digital Twin Registry.

• Extensions

  • Allows you to extend the SDK tool box with your use case specifics and reusable components.

For a full arc42 architecture documentation see docs/architecture/.

For operator and deployment guidance see the Administrator's Guide.

Industry Core Hub Example

You can use it to build you use case application how you want, based on the industry core foundation or not:

Dataspace Architecture Patterns

This SDK will be developed based and will follow the Dataspace & Industry Usage Patterns recommended in Eclipse Tractus-X sig-architecture.

How to Get Involved

• Get onboarded: Getting started. Join the Eclipse Tractus-X open source community as a contributor!
• Read our Contributing Guidelines before submitting pull requests
• Attend the official community office hours and raise your issue!
• Attend our Tractus-X SDK Weekly
• Join our Industry Core Hub Matrix Chat

Found a bug?

👀 If you have identified a bug or want to fix an existing documentation, feel free to create a new issue at our project's corresponding GitHub Issues page

⁉️ Before doing so, please consider searching for potentially suitable existing issues.

🙋 Assign to yourself - Show others that you are working on this issue by assigning it to yourself.
To do so, click the cog wheel next to the Assignees section just to the right of this issue.

Discuss

📣 If you want to share an idea to further enhance the project, please feel free to contribute to the discussions, otherwise create a new discussion

Reporting a Security Issue

Please follow the Security Issue Reporting Guidelines if you come across any security vulnerabilities or concerns.

Contact

• Matrix Chat: Industry Core Hub Channel
• Mailing List: tractusx-dev
• Open Meetings: Tractus-X SDK Weekly
• Community Office Hours: Office Hours

Troubleshooting & FAQ

Q: I get authentication errors when connecting to the connector.
A: Ensure you are using the correct API key or OAuth2 credentials. Verify your X-Api-Key header value and that your connector's management API is accessible.

Q: Which connector versions are supported?
A: The SDK supports EDC v0.8.x through v0.11.x. Use dataspace_version="jupiter" for EDC v0.8.x–v0.10.x and dataspace_version="saturn" for EDC v0.11.x (DSP 2025-1).

Q: How do I resolve dependency conflicts during installation?
A: Try installing in a clean virtual environment: python -m venv .venv && source .venv/bin/activate && pip install tractusx-sdk. If conflicts persist, check the issues page for known problems.

Q: Where can I find more detailed usage examples?
A: See the examples/ directory and the documentation site.

Contributing

Please read our Contributing Guidelines before submitting pull requests.

Licenses

• Apache-2.0 for code
• CC-BY-4.0 for non-code

---

Thank you for using Software Development KIT! If you have any questions or need further assistance, please feel free to reach out.