ECMind Blue Client

A Python client library (Python ≥ 3.12) for the Blue server by ECMind. Communication with the server uses a proprietary TCP/RPC binary protocol.

Full documentation: https://ecmind-blue-client.docs.ecmind.ch — searchable API reference, guides, and examples in German and English.

Using an AI coding assistant? Download the skills bundle for Claude Code, Cline, or Cursor to get LLM-optimised context for every public operation.

Installation

Using uv:

uv add ecmind_blue_client

Using pip:

pip install ecmind_blue_client

Available extras:

Deprecation warning

• ecmind_blue_client.com_client is no longer available.
• ecmind_blue_client.soap_client is no longer available.
• TcpClient and TcpPoolClient are deprecated and superseded by SyncPoolClient / AsyncPoolClient.
• Options is deprecated and superseded by the typed per-job option sets in ecmind_blue_client.ecm (ECMInsertOptions, ECMUpdateOptions, ECMUpsertOptions, ECMDeleteOptions, ECMMoveOptions, ECMCopyOptions).

Documentation

The full reference lives at https://ecmind-blue-client.docs.ecmind.ch and covers:

• API reference — every namespace (ecm.dms, ecm.security, ecm.system, ecm.workflow, ecm.db) with signatures, worked examples, and edge cases
• Guides — quickstart, migration from the deprecated TcpClient/Client, model generation
• Skills bundle — the skills.zip for AI coding assistants. Extract into your tool's skills directory:

  unzip skills.zip -d ~/.claude/skills/
  

The docs are versioned — every released tag has its own browsable copy. The README below is intentionally short and shows only the most common patterns.

High-Level ECM API (ecm/)

The recommended way to interact with the ECM. The entire API is available as both synchronous and asynchronous variants. The ECM() factory function returns either ECMSync or ECMAsync depending on the client type passed.

Setup

from ecmind_blue_client.pool import SyncPoolClient, ServerConnectionSettings
from ecmind_blue_client.ecm import ECM

client = SyncPoolClient(
    servers=[ServerConnectionSettings(hostname="<host>", port=4000)],
    username="<username>",
    password="<password>",
    name="MyApp",
)

ecm = ECM(client)

servers also accepts a compact string in the form "<host>:<port>:<weight>" (multiple servers separated by commas), which is convenient for environment variables:

client = SyncPoolClient(servers="host1:4000:2,host2:4000:1", username="<username>", password="<password>")

For async code, use AsyncPoolClient instead — ECM() will then return an ECMAsync instance with identical await-based methods.

Object related operations (ecm.dms)

Accessed via ecm.dms, this namespace covers all operations on folders, registers, and documents.

Object types can be defined as typed model classes (recommended) or created generically at runtime using the factory functions make_folder_model, make_register_model, and make_document_model — useful when the object type is only known at runtime.

Typed model classes bring a practical advantage in day-to-day development: because all fields are declared as typed attributes, IDEs such as VS Code offer full code completion for field names and their expected data types. More importantly, when the ECM object definition changes — for example when an internal field name is renamed on the server — regenerating the model class causes all affected references across the codebase to be immediately flagged by the IDE or type checker. This makes it straightforward to locate and update every call site without relying on text search.

Model generator

Typed model classes can be generated automatically from a live server or a local asobjdef XML file using the ecm-generate-models command, which is installed alongside the package:

# Generate from a live server
ecm-generate-models --host <host> --username <username> --password <password> --output-dir ./models

# Generate from a local asobjdef XML file
ecm-generate-models --file asobjdef.xml --output-dir ./models

# Generate only a specific cabinet
ecm-generate-models --host <host> --username <username> --password <password> --cabinet MyCabinet --output-dir ./models

# Print to stdout instead of writing files
ecm-generate-models --host <host> --username <username> --password <password>

Replace <host>, <username> and <password> with the actual hostname and login credentials for the target server. Each cabinet produces one .py file in --output-dir containing ready-to-use model classes. SSL is enabled by default; use --no-ssl to disable it. The default port is 4000.

Typed model class (recommended) — definition:

from ecmind_blue_client.ecm.model import ECMFolderModel, ECMField, ECMTableField, ECMTableRowModel

class InvoiceRow(ECMTableRowModel):
    Amount: ECMField[float]
    Description: ECMField[str]

class InvoiceFolder(ECMFolderModel):
    _internal_name_ = "InvoiceFolder"
    Title: ECMField[str]
    Year: ECMField[int]
    Positions: ECMTableField[InvoiceRow]

Typed model class — query with where clauses:

results = ecm.dms.select(InvoiceFolder).where(
    InvoiceFolder.Title == "Invoice 2024",
    (InvoiceFolder.Year >= 2020) & (InvoiceFolder.Year <= 2024),
).order_by(InvoiceFolder.Year.DESC).execute()

for folder in results:
    print(folder.system.id, folder.Title, folder.Year)

Generic model — definition:

from ecmind_blue_client.ecm.model import make_folder_model

InvoiceFolder = make_folder_model("InvoiceFolder")

Generic model — query with where clauses:

results = ecm.dms.select(InvoiceFolder).where(
    InvoiceFolder["Title"] == "Invoice 2024",
    InvoiceFolder["Year"] >= 2020,
).execute()

for folder in results:
    print(folder.system.id, folder["Title"], folder["Year"])

Querying document search flags:

Document base-parameter states such as in a register, signed, without pages or archivable are not normal index fields. Filter on them through the boolean flag properties on a document model's system namespace, used like any other where() condition. Multiple flag conditions combine into a single OBJECT_SEARCHFLAGS condition (the server OR-s flags of the same group and AND-s flags of different groups).

# Documents stored in a register
ecm.dms.select(Invoice).where(Invoice.system.in_register == True).execute()

# Signed documents without pages, narrowed further by a normal field condition
ecm.dms.select(Invoice).where(
    Invoice.system.signed_current == True,
    Invoice.system.without_pages == True,
    Invoice.Year >= 2024,
).execute()

# Paired flags also accept == False (here: documents not in a register)
ecm.dms.select(Invoice).where(Invoice.system.in_register == False).execute()

Available flags: archived, archivable (paired), without_pages, checked_out_by_me, checked_out_by_other, in_register (paired), external, link, multi_location, has_variants, signed_current, signed_former.

Inserting and updating objects:

# Insert and immediately retrieve the created object
folder = ecm.dms.insert_and_get(InvoiceFolder(Title="Invoice 2024", Year=2024))
print(folder.system.id)

# Update an existing object by its system ID
folder.Title = "Updated Title"
ecm.dms.update(folder)

Upsert (insert-or-update):

object_id, type_id, hits, action = (
    ecm.dms.upsert(InvoiceFolder(Title="Invoice 2024", Year=2024))
    .search(InvoiceFolder.Title == "Invoice 2024")
    .execute()
)

Deleting objects:

ecm.dms.delete(folder)

Streaming large result sets:

for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.Title)

# Async variant
async for folder in ecm.dms.select(InvoiceFolder).stream():
    print(folder.Title)

Security operations (ecm.security)

User and group management is accessed via ecm.security.

Reading users, groups, and roles:

# Roles of the currently logged-in user
roles = ecm.security.roles()

# All users (with optional group memberships)
users = ecm.security.users(extended_info=True)

# Detailed attributes for a single user
attrs = ecm.security.user("john")

# All groups, or a single group, or its members
groups = ecm.security.groups()
group = ecm.security.group("Editors")
members = ecm.security.group_members("Editors")

# Groups a specific user belongs to
user_groups = ecm.security.user_groups(user_guid="<guid>")

Creating, updating, and deleting users:

# Create — only `username` is required; password is auto-encoded
new_user = ecm.security.create_user("john", password="S3cret!", display_name="John Doe", email="john@example.com")

# Update — only the named fields are changed; the rest is preserved
ecm.security.update_user(new_user.guid, locked=True)

# Delete — `target_user_guid` is required even when nothing is forwarded
ecm.security.delete_user(new_user.guid, target_user_guid="<admin-guid>")

Group membership:

ecm.security.add_user_to_group(user_guid, group_guid)
ecm.security.remove_user_from_group(user_guid, group_guid)
ecm.security.remove_user_from_all_groups(user_guid)

Checking whether a user can log in (credentials, lock state, password expiry):

from ecmind_blue_client.ecm import ECMUserAccountStatus

check = ecm.security.check_user_account("john", "S3cret!")
if check.login_possible:
    print(f"{check.username} via {check.login_method}, expires in {check.password_expires_in_days} days")
elif check.status is ECMUserAccountStatus.WRONG_PASSWORD:
    print("wrong password")

A wrong password, an unknown user and a locked account come back as a status, not as an exception. The session user needs the SERVER_SWITCH_JOB_CONTEXT system role, and failed attempts count towards the account lockout, exactly like a real login.

Checking a password against the server's password rule (Login\PwdComplexity):

# True when the password satisfies the configured rule; with no rule configured
# (the default) every non-empty password passes.
if not ecm.security.check_password_complexity("S3cret!"):
    raise ValueError("password does not satisfy the server's password rule")

This is not a credential check — the password is never compared against a stored password, so it cannot confirm a user's identity.

Exporting the security system (per-group rights and permission clauses):

export = ecm.security.export_security_system()
for clause in export.group_clauses:
    if clause.delete_clause:
        print(clause.group_name, clause.object_type_name, clause.delete_clause)

System operations (ecm.system)

Server-wide metadata and object definitions are accessed via ecm.system:

# Parsed asobjdef — cabinets, object types, fields (cached after first call)
definition = ecm.system.definition()

# Live snapshot of pool connection statistics
for stats in ecm.system.info():
    print(stats.host, stats.in_use, stats.pool_size)

# License and module information
licenses = ecm.system.check_license("workflow", "archive")
module = ecm.system.module_info("workflow")

# Per-user data — name plus type identify the record
from ecmind_blue_client.ecm import ECMUserDataType

SLOT = ECMUserDataType.ADDITIONAL_APP_CONFIG_80
ecm.system.set_user_data("my_app.last_run", SLOT, b"2026-05-02T10:00:00")
value = ecm.system.get_user_data("my_app.last_run", SLOT)          # bytes | None
names = ecm.system.get_user_data_names(SLOT)                       # upper-cased names
ecm.system.delete_user_data("my_app.last_run", SLOT)               # removes the row

Workflow operations (ecm.workflow)

Workflow organisations, substitutes, and absences are accessed via ecm.workflow:

# All organisations / the active one
orgs = ecm.workflow.organisations()
active = ecm.workflow.active_organisation()

# Objects in the organisation tree (users, roles, groups)
objects = ecm.workflow.organisation_objects(active)

# Mark a user as absent and configure substitutes
ecm.workflow.configure_user_absence(active, {"<user-guid>": True})
ecm.workflow.set_substitutes(active, {"<user-guid>": ["<substitute-guid>"]})

# List currently absent users
absent = ecm.workflow.absent_users(active)

Database access (ecm.db)

Direct SQL queries against the configured ECM database via ado.ExecuteSQL. Placeholders (%s, %w, %d, %f, %u, %%) are bound positionally and prevent SQL injection — never use string formatting:

result = ecm.db.select(
    "SELECT benutzer, osemail FROM benutzer WHERE benutzer = %s",
    "ROOT",
)
for row in result:
    print(row["benutzer"], row["osemail"])

User impersonation

impersonate executes subsequent requests in the security context of another user. All operations performed on the returned instance are treated by the server as if that user had issued them directly — applied rights, audit trail entries, and access restrictions all reflect the target user rather than the authenticated connection user.

The connecting user must hold the system role SERVER_SWITCH_JOB_CONTEXT (ECMSystemRole.SERVER_SWITCH_JOB_CONTEXT, role ID 72) for the server to accept the context switch. Without this role the server will reject the request with an error.

with ecm.impersonate("john") as ecm_john:
    folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

The instance can also be used without a with block when no automatic cleanup is needed:

ecm_john = ecm.impersonate("john")
folder = ecm_john.dms.insert_and_get(InvoiceFolder(Title="Test"))

The target user can be identified by name or GUID; each maps to its own server parameter ($$$SwitchContextUserName$$$, $$$SwitchContextUserGUID$$$). Exactly one of the two must be given — a str is always a user name, never a GUID, because user names may consist of digits or look like a GUID.

There is no identification by numeric user ID: enaio 12.0 accepts $$$SwitchContextUserID$$$ but ignores it, so the job would silently keep running as the logged-in user.

ecm_john = ecm.impersonate("john")                                        # by name
ecm_john = ecm.impersonate(user_guid="8A1D1F2E4C7B4A9E8F0D3C5B7A9E1D2F")  # by GUID

Low-Level RPC API (rpc/)

The RPC layer provides direct TCP socket access to the Blue server. It is the foundation the high-level ECM API is built on. Use it directly only when you need access to server jobs not yet covered by the ECM API.

Connection and job execution

from ecmind_blue_client.pool import SyncPoolClient, ServerConnectionSettings
from ecmind_blue_client.rpc import Jobs

client = SyncPoolClient(
    servers=[ServerConnectionSettings(hostname="<host>", port=4000)],
    username="<username>",
    password="<password>",
)

result = client.execute(Jobs.KRN_GETSERVERINFO, Flags=0, Info=6)
print(result.get("Value", str))

JobResult

The execute() call returns a JobResult:

Session lifecycle

The pool clients manage the full session lifecycle automatically:

1. krn.SessionAttach — establish session
2. krn.SessionLogin — authenticate
3. ECM operations
4. krn.SessionLogout — close session

SSL/TLS is enabled by default. Pass use_ssl=False or a custom cadata PEM string to SyncPoolClient / AsyncPoolClient to override.

Load balancing

Both SyncPoolClient and AsyncPoolClient support weighted load balancing across multiple servers:

from ecmind_blue_client.pool import SyncPoolClient, ServerConnectionSettings

client = SyncPoolClient(
    servers=[
        ServerConnectionSettings(hostname="server1", port=4000, weight=2),
        ServerConnectionSettings(hostname="server2", port=4000, weight=1),
    ],
    username="<username>",
    password="<password>",
    pool_size=10,
)

Keepalive

Long-running processes behind a firewall, NAT gateway or load balancer can have their idle connections dropped silently. Pass keepalive_interval (seconds) to probe every idle pooled connection with krn.CheckServerConnection at that interval; connections that fail the probe are closed and removed instead of failing the next business call. None (the default) or a value <= 0 disables it and starts no background worker at all.

with SyncPoolClient(
    servers="server1:4000:1",
    username="<username>",
    password="<password>",
    keepalive_interval=300,
) as client:
    ...

close() (sync) and aclose() (async) stop the worker and close every idle connection; both pool clients also work as (async) context managers.

Independent of keepalive, the async pool hooks the transport's connection_lost event, so a pooled connection that the server (or a device in between) closes while idle is removed from the pool at the very moment of loss and the next call transparently gets a fresh one. Keepalive still matters for the losses the kernel cannot see at all — a firewall rule removed, a NAT entry expired — and is the only detection available to the sync pool.

Issues and feedback

Bug reports and questions are tracked at the ECM.community.

License

MIT — see LICENSE for the full text.