ecmind-blue-client 1.0.0a4

A client wrapper for blue

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:

Extra	Description
development	Dev tooling: black, isort, pylint, pyright, pytest, pytest-cov, build, twine
manage	(removed) Previously pulled in ecmind-blue-client-manage
objdef	(removed) Previously pulled in ecmind-blue-client-objdef
portfolio	(removed) Previously pulled in ecmind-blue-client-portfolio
workflow	(removed) Previously pulled in ecmind-blue-client-workflow
tcp	(deprecated) Only required for the deprecated TcpClient and TcpPoolClient

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.

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"])

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="[e-mail]")

# 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)

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 (typed accessors for str / int / bytes / datetime)
ecm.system.set_user_data("my_app.last_run", "2026-05-02T10:00:00")
value = ecm.system.get_user_data("my_app.last_run", str)

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"))

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:

Property	Description
result.get(name, type)	Retrieve a typed output parameter
result.files	List of JobResponseFile output file attachments
result.result_code	Server result code (0 = success)
result.error_messages	Server error string, or None on success

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,
)

Issues and feedback

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

License

MIT — see LICENSE for the full text.