Getting Started



The TagManager class is the primary entry point of the Tag API.

When constructing a TagManager, you can pass an HttpConfiguration (like one retrieved from the HttpConfigurationManager), or let TagManager use the default connection. The default connection depends on your SystemLink Client settings.

With a TagManager object, you can:

  • Query, create, modify, and delete tags from the server

    • Use open() to get a tag’s TagData from the server when you know the tag’s path.

    • Use query() to get a collection of TagData objects based on the tags’ paths, keywords, and/or properties.

    • Use refresh() to update a list of TagData objects with fresh metadata from the server.

    • Use update() to modify the server metadata for a list of tags, using either TagData objects to overwrite the server’s tag data or TagDataUpdate objects to selectively update specific fields.

    • Use delete() to delete one or more tags from the server.

  • Read and write tag values

    • Use read() to get a tag’s current value. Via method parameters, you can also request the timestamp indicating when that value was last written and/or the aggregate data stored for the tag (if the tag’s collect_aggregates attribute is enabled on the server).

    • Use create_writer() to get a BufferedTagWriter that will buffer a set of writes, and automatically send the writes when a given buffer_size is reached or when max_buffer_time has elapsed (or when send_buffered_writes() is called).

  • Get a TagSelection that can help perform several of the above operations on several tags at once

If you have a TagSelection, you can use it to create a TagSubscription that will trigger a tag_changed event any time one of the tags’ values is changed.


Read and write individual tags

 1from datetime import timedelta
 3from systemlink.clients.tag import DataType, TagManager
 5mgr = TagManager()
 6tag ="MyTags.Example Tag", DataType.DOUBLE, create=True)
 8with mgr.create_writer(buffer_size=10, max_buffer_time=timedelta(seconds=3)) as writer:
 9    writer.write(tag.path, tag.data_type, 3.5)
10# Note: Exiting the "with" block automatically calls writer.send_buffered_writes()
12read_result =
13assert read_result is not None
14assert read_result.value == 3.5

Subscribe to tag changes

 1from contextlib import ExitStack
 2from time import sleep
 4from systemlink.clients.tag import DataType, TagData, TagManager, TagValueReader
 9def on_tag_changed(tag: TagData, reader: TagValueReader) -> None:
10    """Callback for tag_changed events."""
11    path = tag.path
12    data_type =
14    if reader is not None:
15        read_result =
16        # A read_result of None means that the tag has no value, but it *must*
17        # have a value, because we got a tag_changed event!
18        assert read_result is not None
19        value = read_result.value
20    else:
21        value = "???"  # tag has unknown data type
23    print(f'Tag changed: "{path}" = {value} ({data_type})')
26mgr = TagManager()
28"MyTags.Example Tag", DataType.DOUBLE, create=True)
29    writer = mgr.create_writer(buffer_size=1)
31with ExitStack() as stack:
32    # Notes:
33    # 1. The tags are assumed to already exist before this example is run, but
34    #    setting SIMULATE_EXTERNAL_TAG_CHANGES to True will ensure there is one.
35    # 2. Any tags that get added later will NOT automatically appear in the
36    #    selection just because the path matches the wildcard used below; you
37    #    must call one of the selection's refresh methods to update the tag list
38    #    from the server. But even if you do that:
39    # 3. The subscription will only contain the tags that were in the selection
40    #    when the subscription was created. If you want the subscription to add
41    #    new tags that were added to the selection, you must recreate it.
42    paths = ["MyTags.*"]
43    selection = stack.enter_context(mgr.open_selection(paths))
44    if not selection.metadata:
45        print(f"Found no tags that match {paths}")
46    else:
47        print("Matching tags:")
48        for path in selection.metadata.keys():
49            print(f" - {path}")
50        print()
52        subscription = stack.enter_context(selection.create_subscription())
53        subscription.tag_changed += on_tag_changed
55        # Wait forever, until a KeyboardInterrupt (Ctrl+C)
56        print("Watching for tag changes; hit Ctrl+C to stop")
57        try:
58            i = 0
59            while True:
60                sleep(1)
62                    writer.write("MyTags.Example Tag", DataType.DOUBLE, i)
63                    i += 1
64        except KeyboardInterrupt:
65            pass