Skip to content

Client Reference

Main lib entry point.

teams_lib_pzsp2_z1.client.TeamsClient

The main entry point for interacting with the Microsoft Teams library.

This class manages the lifecycle of the underlying Go subprocess, handles Inter-Process Communication (IPC) via JSON over stdin/stdout, and exposes high-level services for Teams, Channels, and Chats.

Architecture: The Python library acts as a frontend wrapper. It spawns a compiled Go binary (teamsClientLib) as a subprocess. Commands are serialized to JSON and sent to the Go process, which executes the actual Microsoft Graph API calls and returns the results.

Attributes:

Name Type Description
channels ChannelsService

Service for managing channels.
teams TeamsService

Service for managing teams.
chats ChatsService

Service for managing chats and messages.
Source code in teams_lib_pzsp2_z1/client.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class TeamsClient:
    """The main entry point for interacting with the Microsoft Teams library.

    This class manages the lifecycle of the underlying Go subprocess, handles
    Inter-Process Communication (IPC) via JSON over stdin/stdout, and exposes
    high-level services for Teams, Channels, and Chats.

    **Architecture:**
    The Python library acts as a frontend wrapper. It spawns a compiled Go binary
    (`teamsClientLib`) as a subprocess. Commands are serialized to JSON and sent
    to the Go process, which executes the actual Microsoft Graph API calls and
    returns the results.

    Attributes:
        channels (ChannelsService): Service for managing channels.
        teams (TeamsService): Service for managing teams.
        chats (ChatsService): Service for managing chats and messages.
    """


    def __init__(
        self,
        auto_init: bool = True,
        env_path: str | None = None,
        cache_mode: config.CacheMode = config.CacheMode.DISABLED,
        cache_path: str | None = None,
    ):
        """Initializes the TeamsClient and spawns the Go subprocess.

        Args:
            auto_init (bool, optional): If True, automatically calls `init_client`
                using configuration loaded from the environment. Defaults to True.
            env_path (str | None, optional): Path to the `.env` file containing
                credentials (CLIENT_ID, TENANT_ID, etc.). If None, defaults are used.
            cache_mode (config.CacheMode, optional): The caching strategy to use
                (e.g., DISABLED, IN_MEMORY, DISK). Defaults to DISABLED.
            cache_path (str | None, optional): File path for the cache (required if
                cache_mode is DISK). Defaults to None.

        Raises:
            RuntimeError: If the operating system is not supported (only Windows/Linux).
        """

        self._lock = threading.Lock()

        # Spawn the Go binary in a subprocess with piped IO
        self.proc = subprocess.Popen(  # noqa: S603
            [str(self._binary())],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=sys.stderr, # Pass Go logs directly to stderr
            text=True,
            bufsize=1,
        )

        self.env_path = env_path

        # Initialize services with a reference to self (to access .execute())
        self.channels = ChannelsService(self)
        self.teams = TeamsService(self)
        self.chats = ChatsService(self)

        self.is_cache_enabled = cache_mode != config.CacheMode.DISABLED

        if auto_init:
            self.init_client(cache_mode, cache_path)

    def _binary(self):
        """Resolves the path to the correct Go binary for the current OS.

        Returns:
            pathlib.Path: The absolute path to the executable.

        Raises:
            RuntimeError: If the OS is not Windows or Linux.
        """

        base = pathlib.Path(__file__).parent / "bin"
        osname = platform.system()

        if osname == "Windows":
            return base / "teamsClientLib_windows.exe"
        elif osname == "Linux":
            return base / "teamsClientLib_linux"
        else:
            raise RuntimeError("Unsupported OS")

    def init_client(
        self,
        cache_mode: config.CacheMode = config.CacheMode.DISABLED,
        cache_path: str | None = None,
    ) -> Any:
        """Initializes the Go backend with authentication and cache configuration.

        This method sends the 'init' command to the Go process, which sets up the
        MSAL token provider and the Graph Service Client.

        Args:
            cache_mode (config.CacheMode): The caching strategy.
            cache_path (str | None): Path to the cache file (if applicable).

        Returns:
            Any: The initialization result from the Go process.

        Raises:
            RuntimeError: If the Go process reports an initialization error.
        """

        sender_config = config.SenderConfig()
        auth_config = config.load_auth_config(self.env_path)
        return self.execute(
            cmd_type="init",
            config={
                "senderConfig": {
                    "maxRetries": sender_config.max_retries,
                    "nextRetryDelay": sender_config.next_retry_delay,
                    "timeout": sender_config.timeout,
                },
                "authConfig": {
                    "clientID": auth_config.client_id,
                    "tenant": auth_config.tenant,
                    "email": auth_config.email,
                    "scopes": auth_config.scopes,
                    "authMethod": auth_config.auth_method,
                },
                "cacheMode": cache_mode.value,
                "cachePath": cache_path,
            },
        )

    def init_fake_client(self, mock_server_url: str) -> Any:
        """Initializes the Go backend in test mode using a mock server.

        This bypasses real MSAL authentication and directs Graph API calls to
        the provided local URL. [For testing purposes only]

        Args:
            mock_server_url (str): The URL of the mock HTTP server.

        Returns:
            Any: The result of the initialization.
        """

        return self.execute(
            cmd_type="init",
            params={
                "mockServerUrl": mock_server_url,
            },
        )

    def execute(
        self,
        cmd_type: str,
        method: str | None = None,
        config: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        """Executes a command on the Go subprocess via JSON-IPC.

        This is the low-level bridge method used by services to communicate with the backend.
        It handles serialization, thread-safe writing to stdin, reading from stdout,
        and error propagation.

        Args:
            cmd_type (str): The type of command (e.g., "init", "request").
            method (str | None): The specific API method to call (e.g., "listChannels").
                Required if cmd_type is "request".
            config (dict | None): Configuration payload (used primarily for initialization).
            params (dict | None): Parameters for the method call (e.g., teamRef, body).

        Returns:
            Any: The 'result' field from the Go response.

        Raises:
            RuntimeError: If the Go process crashes, closes the pipe, returns an empty response,
                or explicitly reports an error in the "error" field.
        """

        payload = {"type": cmd_type}
        if method:
            payload["method"] = method
        if params:
            payload["params"] = params
        if config:
            payload["config"] = config

        json_payload = json.dumps(payload)

        # Critical section to avoid interleaving requests/responses#
        # (e.g., thread A writes request A, thread B writes request B,
        # then thread A reads response B).
        with self._lock:
            try:
                self.proc.stdin.write(json_payload + "\n")
                self.proc.stdin.flush()

                raw_response = self.proc.stdout.readline()

            except BrokenPipeError:
                raise RuntimeError("Go process crashed or closed connection")  # noqa: B904

            if not raw_response:
                raise RuntimeError("Go process returned empty response")

            res = json.loads(raw_response)

        if "error" in res and res["error"]:
            raise RuntimeError(f"Go Error: {res['error']}")

        return res.get("result")

    def close(self):
        """Gracefully closes the Go backend, ensuring cache is synced.

        Sends a 'close' command to the Go process, which triggers `lib.Close()`
        to wait for background cache operations. Then terminates the process.
        """

        if self.is_cache_enabled:
            with self._lock:
                if self.proc.poll() is None:
                    try:
                        payload = json.dumps({"type": "close"}) + "\n"
                        self.proc.stdin.write(payload)
                        self.proc.stdin.flush()
                    except (BrokenPipeError, OSError):
                        pass

            try:
                self.proc.wait(timeout=3)
            except subprocess.TimeoutExpired:
                self.proc.kill()
                self.proc.wait()

            try:
                if self.proc.stdin:
                    self.proc.stdin.close()
                if self.proc.stdout:
                    self.proc.stdout.close()
            except (OSError, ValueError):
                pass
        else:
            self.proc.terminate()
            self.proc.wait()

Functions

__init__(auto_init=True, env_path=None, cache_mode=config.CacheMode.DISABLED, cache_path=None)

Initializes the TeamsClient and spawns the Go subprocess.

Parameters:

Name Type Description Default
auto_init bool

If True, automatically calls init_client using configuration loaded from the environment. Defaults to True.

 True
env_path str | None

Path to the .env file containing credentials (CLIENT_ID, TENANT_ID, etc.). If None, defaults are used.

 None
cache_mode CacheMode

The caching strategy to use (e.g., DISABLED, IN_MEMORY, DISK). Defaults to DISABLED.

 DISABLED
cache_path str | None

File path for the cache (required if cache_mode is DISK). Defaults to None.

 None

Raises:

Type Description
RuntimeError

If the operating system is not supported (only Windows/Linux).
Source code in teams_lib_pzsp2_z1/client.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(
    self,
    auto_init: bool = True,
    env_path: str | None = None,
    cache_mode: config.CacheMode = config.CacheMode.DISABLED,
    cache_path: str | None = None,
):
    """Initializes the TeamsClient and spawns the Go subprocess.

    Args:
        auto_init (bool, optional): If True, automatically calls `init_client`
            using configuration loaded from the environment. Defaults to True.
        env_path (str | None, optional): Path to the `.env` file containing
            credentials (CLIENT_ID, TENANT_ID, etc.). If None, defaults are used.
        cache_mode (config.CacheMode, optional): The caching strategy to use
            (e.g., DISABLED, IN_MEMORY, DISK). Defaults to DISABLED.
        cache_path (str | None, optional): File path for the cache (required if
            cache_mode is DISK). Defaults to None.

    Raises:
        RuntimeError: If the operating system is not supported (only Windows/Linux).
    """

    self._lock = threading.Lock()

    # Spawn the Go binary in a subprocess with piped IO
    self.proc = subprocess.Popen(  # noqa: S603
        [str(self._binary())],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=sys.stderr, # Pass Go logs directly to stderr
        text=True,
        bufsize=1,
    )

    self.env_path = env_path

    # Initialize services with a reference to self (to access .execute())
    self.channels = ChannelsService(self)
    self.teams = TeamsService(self)
    self.chats = ChatsService(self)

    self.is_cache_enabled = cache_mode != config.CacheMode.DISABLED

    if auto_init:
        self.init_client(cache_mode, cache_path)

close()

Gracefully closes the Go backend, ensuring cache is synced.

Sends a 'close' command to the Go process, which triggers lib.Close() to wait for background cache operations. Then terminates the process.

Source code in teams_lib_pzsp2_z1/client.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def close(self):
    """Gracefully closes the Go backend, ensuring cache is synced.

    Sends a 'close' command to the Go process, which triggers `lib.Close()`
    to wait for background cache operations. Then terminates the process.
    """

    if self.is_cache_enabled:
        with self._lock:
            if self.proc.poll() is None:
                try:
                    payload = json.dumps({"type": "close"}) + "\n"
                    self.proc.stdin.write(payload)
                    self.proc.stdin.flush()
                except (BrokenPipeError, OSError):
                    pass

        try:
            self.proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            self.proc.kill()
            self.proc.wait()

        try:
            if self.proc.stdin:
                self.proc.stdin.close()
            if self.proc.stdout:
                self.proc.stdout.close()
        except (OSError, ValueError):
            pass
    else:
        self.proc.terminate()
        self.proc.wait()

execute(cmd_type, method=None, config=None, params=None)

Executes a command on the Go subprocess via JSON-IPC.

This is the low-level bridge method used by services to communicate with the backend. It handles serialization, thread-safe writing to stdin, reading from stdout, and error propagation.

Parameters:

Name Type Description Default
cmd_type str

The type of command (e.g., "init", "request").

 required
method str | None

The specific API method to call (e.g., "listChannels"). Required if cmd_type is "request".

 None
config dict | None

Configuration payload (used primarily for initialization).

 None
params dict | None

Parameters for the method call (e.g., teamRef, body).

 None

Returns:

Name Type Description
Any Any

The 'result' field from the Go response.

Raises:

Type Description
RuntimeError

If the Go process crashes, closes the pipe, returns an empty response, or explicitly reports an error in the "error" field.
Source code in teams_lib_pzsp2_z1/client.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def execute(
    self,
    cmd_type: str,
    method: str | None = None,
    config: dict[str, Any] | None = None,
    params: dict[str, Any] | None = None,
) -> Any:
    """Executes a command on the Go subprocess via JSON-IPC.

    This is the low-level bridge method used by services to communicate with the backend.
    It handles serialization, thread-safe writing to stdin, reading from stdout,
    and error propagation.

    Args:
        cmd_type (str): The type of command (e.g., "init", "request").
        method (str | None): The specific API method to call (e.g., "listChannels").
            Required if cmd_type is "request".
        config (dict | None): Configuration payload (used primarily for initialization).
        params (dict | None): Parameters for the method call (e.g., teamRef, body).

    Returns:
        Any: The 'result' field from the Go response.

    Raises:
        RuntimeError: If the Go process crashes, closes the pipe, returns an empty response,
            or explicitly reports an error in the "error" field.
    """

    payload = {"type": cmd_type}
    if method:
        payload["method"] = method
    if params:
        payload["params"] = params
    if config:
        payload["config"] = config

    json_payload = json.dumps(payload)

    # Critical section to avoid interleaving requests/responses#
    # (e.g., thread A writes request A, thread B writes request B,
    # then thread A reads response B).
    with self._lock:
        try:
            self.proc.stdin.write(json_payload + "\n")
            self.proc.stdin.flush()

            raw_response = self.proc.stdout.readline()

        except BrokenPipeError:
            raise RuntimeError("Go process crashed or closed connection")  # noqa: B904

        if not raw_response:
            raise RuntimeError("Go process returned empty response")

        res = json.loads(raw_response)

    if "error" in res and res["error"]:
        raise RuntimeError(f"Go Error: {res['error']}")

    return res.get("result")

init_client(cache_mode=config.CacheMode.DISABLED, cache_path=None)

Initializes the Go backend with authentication and cache configuration.

This method sends the 'init' command to the Go process, which sets up the MSAL token provider and the Graph Service Client.

Parameters:

Name Type Description Default
cache_mode CacheMode

The caching strategy.

 DISABLED
cache_path str | None

Path to the cache file (if applicable).

 None

Returns:

Name Type Description
Any Any

The initialization result from the Go process.

Raises:

Type Description
RuntimeError

If the Go process reports an initialization error.
Source code in teams_lib_pzsp2_z1/client.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def init_client(
    self,
    cache_mode: config.CacheMode = config.CacheMode.DISABLED,
    cache_path: str | None = None,
) -> Any:
    """Initializes the Go backend with authentication and cache configuration.

    This method sends the 'init' command to the Go process, which sets up the
    MSAL token provider and the Graph Service Client.

    Args:
        cache_mode (config.CacheMode): The caching strategy.
        cache_path (str | None): Path to the cache file (if applicable).

    Returns:
        Any: The initialization result from the Go process.

    Raises:
        RuntimeError: If the Go process reports an initialization error.
    """

    sender_config = config.SenderConfig()
    auth_config = config.load_auth_config(self.env_path)
    return self.execute(
        cmd_type="init",
        config={
            "senderConfig": {
                "maxRetries": sender_config.max_retries,
                "nextRetryDelay": sender_config.next_retry_delay,
                "timeout": sender_config.timeout,
            },
            "authConfig": {
                "clientID": auth_config.client_id,
                "tenant": auth_config.tenant,
                "email": auth_config.email,
                "scopes": auth_config.scopes,
                "authMethod": auth_config.auth_method,
            },
            "cacheMode": cache_mode.value,
            "cachePath": cache_path,
        },
    )

init_fake_client(mock_server_url)

Initializes the Go backend in test mode using a mock server.

This bypasses real MSAL authentication and directs Graph API calls to the provided local URL. [For testing purposes only]

Parameters:

Name Type Description Default
mock_server_url str

The URL of the mock HTTP server.

 required

Returns:

Name Type Description
Any Any

The result of the initialization.
Source code in teams_lib_pzsp2_z1/client.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def init_fake_client(self, mock_server_url: str) -> Any:
    """Initializes the Go backend in test mode using a mock server.

    This bypasses real MSAL authentication and directs Graph API calls to
    the provided local URL. [For testing purposes only]

    Args:
        mock_server_url (str): The URL of the mock HTTP server.

    Returns:
        Any: The result of the initialization.
    """

    return self.execute(
        cmd_type="init",
        params={
            "mockServerUrl": mock_server_url,
        },
    )