inspection
Shared runtime inspection and JSON serialization helpers.
func_sdk_version() -> strResolve the installed SDK version lazily.
Returns
strfuncdescribe_local_server() -> strSummarize the state of the local Kitaru-compatible server, if any.
Returns
strfuncconnected_to_local_server_safe() -> boolSafely check whether the current client is bound to a local server.
Returns
boolfunc_build_snapshot_without_local_store(_gc, _exc) -> RuntimeSnapshotBuild a degraded snapshot when local runtime support is unavailable.
param_gcGlobalConfigurationparam_excExceptionReturns
kitaru.inspection.RuntimeSnapshotfuncuses_stale_local_server_url(server_url, local_server_status) -> boolCheck for a localhost URL that points at a stopped local server.
paramserver_urlstr | Noneparamlocal_server_statusstr | NoneReturns
boolfunc_legacy_runner_env_warning() -> str | NoneReturn a warning when the legacy stack-selection env var is still set.
Returns
str | Nonefunclog_store_mismatch_details(preferred) -> tuple[str | None, str | None]Return status-row + warning text when preferred and active backends differ.
parampreferredResolvedLogStoreReturns
tuple[str | None, str | None]funccombine_warnings(*warnings) -> str | NoneCombine non-empty warning messages into one multiline block.
paramwarningsstr | None= ()Returns
str | Nonefuncbuild_runtime_snapshot() -> RuntimeSnapshotResolve the current Kitaru runtime state from ZenML-backed config.
Returns
kitaru.inspection.RuntimeSnapshotfunc_qualified_type_name(value) -> strReturn the fully qualified runtime type name for a value.
paramvalueAnyReturns
strfuncto_jsonable(value, *, fallback_repr) -> AnyConvert a value into a JSON-serializable representation.
paramvalueAnyparamfallback_reprboolReturns
typing.Anyfuncserialize_failure(failure) -> dict[str, Any] | NoneSerialize optional failure details.
paramfailureFailureInfo | NoneReturns
dict[str, typing.Any] | Nonefuncserialize_pending_wait(wait) -> dict[str, Any] | NoneSerialize optional pending wait details.
paramwaitPendingWait | NoneReturns
dict[str, typing.Any] | Nonefuncserialize_artifact_ref(artifact) -> dict[str, Any]Serialize artifact metadata.
paramartifactArtifactRefReturns
dict[str, typing.Any]funcserialize_artifact_value(value) -> dict[str, Any]Serialize an artifact payload value for MCP transport.
paramvalueAnyReturns
dict[str, typing.Any]funcserialize_checkpoint_attempt(attempt) -> dict[str, Any]Serialize checkpoint-attempt details.
paramattemptCheckpointAttemptReturns
dict[str, typing.Any]funcserialize_checkpoint_call(checkpoint) -> dict[str, Any]Serialize checkpoint-call details.
paramcheckpointCheckpointCallReturns
dict[str, typing.Any]funcserialize_execution_summary(execution) -> dict[str, Any]Serialize execution list-item details.
paramexecutionExecutionReturns
dict[str, typing.Any]funcserialize_execution(execution) -> dict[str, Any]Serialize full execution details.
paramexecutionExecutionReturns
dict[str, typing.Any]funcserialize_stack(stack, *, is_managed=None) -> dict[str, Any]Serialize stack information for structured output.
paramstackStackInfoparamis_managedbool | None= NoneReturns
dict[str, typing.Any]funcserialize_stack_create_result(result) -> dict[str, Any]Serialize stack-create operation details.
paramresult_StackCreateResultReturns
dict[str, typing.Any]funcserialize_stack_delete_result(result) -> dict[str, Any]Serialize stack-delete operation details.
paramresult_StackDeleteResultReturns
dict[str, typing.Any]func_serialize_stack_component_details(component) -> dict[str, Any]Serialize one translated stack component for structured stack output.
paramcomponentStackComponentDetailsReturns
dict[str, typing.Any]funcserialize_stack_details(details) -> dict[str, Any]Serialize stack inspection details for stack show style output.
paramdetailsStackDetailsReturns
dict[str, typing.Any]funcserialize_runtime_snapshot(snapshot) -> dict[str, Any]Serialize runtime status details for structured output.
paramsnapshotRuntimeSnapshotReturns
dict[str, typing.Any]funcserialize_log_entry(entry) -> dict[str, Any]Serialize one log entry for JSON output.
paramentryLogEntryReturns
dict[str, typing.Any]funcserialize_model_alias(entry) -> dict[str, Any]Serialize model alias information.
paramentryModelAliasEntryReturns
dict[str, typing.Any]funcsecret_visibility(secret) -> strReturn a human-readable visibility label for a secret.
paramsecretSecretResponseReturns
strfuncserialize_secret_summary(secret) -> dict[str, Any]Serialize secret summary information.
paramsecretSecretResponseReturns
dict[str, typing.Any]funcserialize_secret_detail(secret, *, show_values) -> dict[str, Any]Serialize secret detail information.
paramsecretSecretResponseparamshow_valuesboolReturns
dict[str, typing.Any]funcserialize_resolved_log_store(snapshot, *, active_store=None, warning=None) -> dict[str, Any]Serialize effective log-store information.
paramsnapshotResolvedLogStoreparamactive_storeActiveStackLogStore | None= Noneparamwarningstr | None= NoneReturns
dict[str, typing.Any]