name: kelvin-sdk-importer
description: Use when implementing, reviewing, debugging, refactoring, or migrating Kelvin importer applications (type: importer) with the Kelvin Python SDK, including importer app.yaml structure, importer_io declarations, io_configuration UI schemas, manual async ingestion loops, runtime-mapped stream configuration, custom action inputs/outputs and on_custom_action handling, external connector patterns (MQTT, OPC-UA, REST, etc.), and KRN construction/parsing. Do NOT use for SmartApps (type: app); use kelvin-sdk-app instead.
Kelvin SDK — Importer Developer
Build and modify Kelvin importer applications (type: importer) with the Kelvin Python SDK. Importers ingest data from external systems (MQTT, OPC-UA, REST, databases, files, etc.) and publish it to the Kelvin Platform. Start with a minimal working implementation, then expand only for explicit requirements.
Execution Workflow
- Clarify only missing requirements.
- Identify connector type (MQTT, OPC-UA, REST, file, database, etc.) and data types.
- Choose the first reference file with deterministic decision rules.
- Load only additional references required by the task.
- Implement with the manual async loop runtime model.
- Validate against the rule checklist before finalizing.
Clarify Missing Requirements
Ask at most 2-3 high-impact questions at a time, and only when missing information blocks correct implementation.
Prioritize questions in this order:
- External source: protocol, connection details, and data format.
- Data types: what kinds of streams will be published (number, string, boolean, object).
- Control: whether the importer handles incoming control changes from the platform.
- Custom actions: whether the importer receives custom actions from the platform (e.g. trigger a remote command on the external system) or emits result actions back.
- Importer mapping: per-stream IO configuration shape (what the operator configures per mapped stream).
- Configuration: app-level connector settings (host, port, polling interval, etc.).
If the request is already explicit enough, proceed without extra questions.
When details are missing but non-blocking, proceed with explicit assumptions and mark them clearly:
- Use placeholder names and configuration fields when concrete details are unknown.
- Prefer configurable values (app configuration) over hardcoded constants.
- Do not invent credentials, KRN identifiers, or environment-specific secrets.
- Summarize all assumptions in the final response so they can be confirmed quickly.
First-File Decision Rules
Pick exactly one first reference file from the list below, then expand only if needed:
Do not load all references by default. Load only what the current task needs.
Implementation Defaults
- Use a manual async loop with
await app.connect() — importers own the event loop.
- Do not use
@app.stream(), @app.timer(), or @app.task as the primary ingestion mechanism.
- Build routing from runtime asset/datastream configuration instead of hardcoding asset or stream bindings.
- Read per-stream mapping from
app.assets[asset].datastreams[stream].configuration.
- Re-read
app.app_configuration inside reconnect/polling loops so runtime updates take effect.
- Use
importer_io and ui_schemas.io_configuration — never use data_streams.inputs for external signals.
- Catch both connector-library exceptions and
OSError for network failures in reconnect loops.
- For custom actions, declare
custom_actions.inputs / custom_actions.outputs at the root of app.yaml (the same schema SmartApps use, not a field of importer_io) and wire app.on_custom_action before await app.connect(). Publish CustomActionResult back to acknowledge.
Validation Checklist
app.yaml and schema alignment
- Use
type: importer, not type: app.
- Use
importer_io to declare ingest capabilities, not data_streams.inputs.
- Keep
ui_schemas.io_configuration.<profile> aligned with importer_io[].name.
- Use
defaults.configuration for connector settings (host, port, polling interval, etc.).
- If the importer handles custom actions, declare them under a top-level
custom_actions: block (with inputs: and/or outputs: listing type: entries) — this is the same schema as SmartApps and lives outside importer_io.
- Do not declare Kelvin runtime credentials (
KELVIN_CLIENT__URL, KELVIN_CLIENT__CLIENT_ID, KELVIN_CLIENT__CLIENT_SECRET) in defaults.system.environment_vars; the platform injects them.
- Follow naming conventions: app names with hyphens, stream/parameter names with underscores.
Runtime and data publishing
- Call
await app.connect() explicitly; do not use app.run().
- Read per-stream mapping from
asset_info.datastreams[stream_name].configuration.
- Publish with
Message and KMessageTypeData, matching the primitive to the mapped stream data type.
- Publish only to mapped asset/stream targets derived from runtime IO configuration.
- If
custom_actions.inputs is declared, assign app.on_custom_action = <async handler> before await app.connect() and respond with CustomActionResult (or publish to a declared custom_actions.outputs type).
Reliability and safety
- Catch connector-library exceptions and
OSError in reconnect loops.
- Log connection state, reconnection attempts, and mapping issues with context.
- Keep secrets and credentials out of source files.
- Do not add broker usernames or passwords unless the external system actually requires them.
Framework Guarantees
Assume these SDK guarantees and avoid redundant checks:
app.assets contains the current set of deployed assets with their datastream configurations.asset_info.datastreams[stream].configuration contains the operator-configured per-stream mapping.app.app_configuration contains the current global app configuration.
Validate connector-specific logic (connection parameters, data parsing, protocol handling), not framework invariants.
