0
# Client Operations
1
2
Client-side functionality for connecting to language servers, handling server responses, and building language client applications with support for multiple transport methods and message handling.
3
4
## Capabilities
5
6
### JSON-RPC Client
7
8
Base client implementation for connecting to JSON-RPC servers with support for multiple transport methods and message handling.
9
10
```python { .api }
11
class JsonRPCClient:
12
"""
13
JSON-RPC client for connecting to language servers.
14
15
Provides client-side JSON-RPC communication with support for stdio,
16
TCP, and WebSocket transports, plus message routing and response handling.
17
"""
18
19
def __init__(
20
self,
21
protocol_cls: Type[JsonRPCProtocol] = None,
22
converter_factory: Callable[[], Converter] = None
23
):
24
"""
25
Initialize JSON-RPC client.
26
27
Parameters:
28
- protocol_cls: Type[JsonRPCProtocol] - Protocol class for communication
29
- converter_factory: Callable - Factory for creating message converters
30
"""
31
32
async def start_io(self, cmd: str, *args, **kwargs) -> None:
33
"""
34
Start server process and communicate over stdio.
35
36
Parameters:
37
- cmd: str - Server command to execute
38
- *args: Additional command arguments
39
- **kwargs: Additional keyword arguments for subprocess
40
"""
41
42
def start_tcp(self, host: str, port: int) -> None:
43
"""
44
Start client with TCP transport.
45
46
Parameters:
47
- host: str - Server host address
48
- port: int - Server port number
49
"""
50
51
def start_ws(self, host: str, port: int) -> None:
52
"""
53
Start client with WebSocket transport.
54
55
Parameters:
56
- host: str - Server host address
57
- port: int - Server port number
58
"""
59
60
def feature(self, method_name: str) -> Callable[[F], F]:
61
"""
62
Decorator for registering notification handlers.
63
64
Parameters:
65
- method_name: str - LSP method name to handle
66
67
Returns:
68
Decorator function for handler registration
69
"""
70
71
72
async def server_exit(self, server: 'asyncio.subprocess.Process') -> None:
73
"""
74
Called when server process exits (overridable).
75
76
Parameters:
77
- server: asyncio.subprocess.Process - Exited server process
78
"""
79
80
def report_server_error(
81
self,
82
error: Exception,
83
source: Union[PyglsError, JsonRpcException]
84
) -> None:
85
"""
86
Report server errors (overridable).
87
88
Parameters:
89
- error: Exception - Error that occurred
90
- source: Union[PyglsError, JsonRpcException] - Error source
91
"""
92
93
async def stop(self) -> None:
94
"""Stop the client and clean up resources."""
95
96
@property
97
def stopped(self) -> bool:
98
"""Whether client has been stopped."""
99
100
@property
101
def protocol(self) -> JsonRPCProtocol:
102
"""Access to underlying protocol instance."""
103
```
104
105
### Base Language Server Client
106
107
Generated LSP client with all standard Language Server Protocol methods for comprehensive server interaction.
108
109
```python { .api }
110
class BaseLanguageClient(JsonRPCClient):
111
"""
112
LSP client with complete Language Server Protocol method support.
113
114
Auto-generated client providing all standard LSP requests and
115
notifications with proper parameter typing and response handling.
116
"""
117
118
# Note: This class contains numerous auto-generated methods
119
# for all LSP features. Key examples include:
120
121
async def text_document_completion_async(
122
self,
123
params: CompletionParams
124
) -> Union[List[CompletionItem], CompletionList, None]:
125
"""Send completion request to server."""
126
127
async def text_document_hover_async(
128
self,
129
params: HoverParams
130
) -> Optional[Hover]:
131
"""Send hover request to server."""
132
133
async def text_document_definition_async(
134
self,
135
params: DefinitionParams
136
) -> Union[Location, List[Location], List[LocationLink], None]:
137
"""Send go-to-definition request to server."""
138
139
def text_document_did_open(self, params: DidOpenTextDocumentParams) -> None:
140
"""Send document open notification."""
141
142
def text_document_did_change(self, params: DidChangeTextDocumentParams) -> None:
143
"""Send document change notification."""
144
145
def text_document_did_close(self, params: DidCloseTextDocumentParams) -> None:
146
"""Send document close notification."""
147
```
148
149
## Usage Examples
150
151
### Basic Client Setup
152
153
```python
154
import asyncio
155
from pygls.client import JsonRPCClient
156
from lsprotocol.types import (
157
InitializeParams,
158
ClientCapabilities,
159
TextDocumentClientCapabilities,
160
CompletionClientCapabilities
161
)
162
163
class LanguageServerClient(JsonRPCClient):
164
def __init__(self):
165
super().__init__()
166
self.server_capabilities = None
167
168
async def initialize_server(self):
169
"""Initialize connection with language server."""
170
# Send initialize request
171
initialize_params = InitializeParams(
172
process_id=os.getpid(),
173
root_uri="file:///path/to/project",
174
capabilities=ClientCapabilities(
175
text_document=TextDocumentClientCapabilities(
176
completion=CompletionClientCapabilities(
177
dynamic_registration=True,
178
completion_item={
179
"snippet_support": True,
180
"documentation_format": ["markdown", "plaintext"]
181
}
182
)
183
)
184
),
185
initialization_options={}
186
)
187
188
result = await self.protocol.send_request(
189
"initialize",
190
initialize_params
191
)
192
193
self.server_capabilities = result.capabilities
194
195
# Send initialized notification
196
self.protocol.send_notification("initialized", {})
197
198
return result
199
200
async def shutdown_server(self):
201
"""Gracefully shutdown server connection."""
202
await self.protocol.send_request("shutdown", None)
203
self.protocol.send_notification("exit", None)
204
205
# Usage
206
async def main():
207
client = LanguageServerClient()
208
209
try:
210
# Start client with stdio to connect to server
211
client.start_io(sys.stdin, sys.stdout)
212
213
# Initialize server
214
init_result = await client.initialize_server()
215
print(f"Server initialized: {init_result.server_info.name}")
216
217
# Use server features...
218
219
finally:
220
await client.shutdown_server()
221
222
asyncio.run(main())
223
```
224
225
### Handling Server Notifications
226
227
```python
228
from pygls.client import JsonRPCClient
229
230
class NotificationHandlingClient(JsonRPCClient):
231
def __init__(self):
232
super().__init__()
233
self.diagnostics = {}
234
235
@self.feature("textDocument/publishDiagnostics")
236
def handle_diagnostics(self, params):
237
"""Handle diagnostic notifications from server."""
238
uri = params.uri
239
diagnostics = params.diagnostics
240
241
self.diagnostics[uri] = diagnostics
242
243
print(f"Received {len(diagnostics)} diagnostics for {uri}")
244
for diagnostic in diagnostics:
245
print(f" {diagnostic.severity}: {diagnostic.message}")
246
247
@self.feature("window/logMessage")
248
def handle_log_message(self, params):
249
"""Handle log messages from server."""
250
print(f"Server log [{params.type}]: {params.message}")
251
252
@self.feature("window/showMessage")
253
def handle_show_message(self, params):
254
"""Handle show message requests from server."""
255
print(f"Server message [{params.type}]: {params.message}")
256
```
257
258
### Interactive Client Operations
259
260
```python
261
import asyncio
262
from lsprotocol.types import (
263
DidOpenTextDocumentParams,
264
TextDocumentItem,
265
CompletionParams,
266
Position,
267
TextDocumentIdentifier
268
)
269
270
class InteractiveClient(JsonRPCClient):
271
async def open_document(self, uri: str, content: str):
272
"""Open a document on the server."""
273
params = DidOpenTextDocumentParams(
274
text_document=TextDocumentItem(
275
uri=uri,
276
language_id="python",
277
version=1,
278
text=content
279
)
280
)
281
282
self.protocol.send_notification("textDocument/didOpen", params)
283
print(f"Opened document: {uri}")
284
285
async def get_completions(self, uri: str, line: int, character: int):
286
"""Request completions at a specific position."""
287
params = CompletionParams(
288
text_document=TextDocumentIdentifier(uri=uri),
289
position=Position(line=line, character=character)
290
)
291
292
result = await self.protocol.send_request(
293
"textDocument/completion",
294
params
295
)
296
297
if result:
298
if isinstance(result, list):
299
return result
300
else:
301
return result.items
302
return []
303
304
async def get_hover(self, uri: str, line: int, character: int):
305
"""Request hover information at a specific position."""
306
params = HoverParams(
307
text_document=TextDocumentIdentifier(uri=uri),
308
position=Position(line=line, character=character)
309
)
310
311
result = await self.protocol.send_request(
312
"textDocument/hover",
313
params
314
)
315
316
return result
317
318
# Interactive usage
319
async def interactive_session():
320
client = InteractiveClient()
321
client.start_tcp("localhost", 8080)
322
323
await client.initialize_server()
324
325
# Open a Python file
326
python_code = '''
327
def hello_world():
328
print("Hello, world!")
329
return "success"
330
331
hello_world().
332
'''
333
334
await client.open_document("file:///test.py", python_code)
335
336
# Get completions after the dot
337
completions = await client.get_completions("file:///test.py", 4, 15)
338
print("Available completions:", [item.label for item in completions])
339
340
# Get hover info for function name
341
hover = await client.get_hover("file:///test.py", 1, 4)
342
if hover:
343
print("Hover content:", hover.contents)
344
345
await client.shutdown_server()
346
347
asyncio.run(interactive_session())
348
```
349
350
### Client with Custom Protocol
351
352
```python
353
from pygls.protocol import JsonRPCProtocol
354
355
class CustomClientProtocol(JsonRPCProtocol):
356
def __init__(self, client, converter):
357
super().__init__(client, converter)
358
self.custom_features = {}
359
360
async def send_custom_request(self, method: str, params: Any):
361
"""Send custom request to server."""
362
return await self.send_request(f"custom/{method}", params)
363
364
def handle_custom_notification(self, method: str, params: Any):
365
"""Handle custom notifications from server."""
366
handler = self.custom_features.get(method)
367
if handler:
368
handler(params)
369
370
class CustomClient(JsonRPCClient):
371
def __init__(self):
372
super().__init__(protocol_cls=CustomClientProtocol)
373
374
def register_custom_feature(self, method: str, handler: Callable):
375
"""Register handler for custom server notifications."""
376
self.protocol.custom_features[method] = handler
377
378
async def call_custom_feature(self, feature: str, params: Any):
379
"""Call custom server feature."""
380
return await self.protocol.send_custom_request(feature, params)
381
382
# Usage with custom protocol
383
client = CustomClient()
384
385
# Register custom notification handler
386
client.register_custom_feature(
387
"analysis_complete",
388
lambda params: print(f"Analysis completed: {params}")
389
)
390
391
# Call custom server feature
392
result = await client.call_custom_feature("analyze_project", {
393
"path": "/path/to/project",
394
"deep_analysis": True
395
})
396
```
397
398
### Error Handling and Reconnection
399
400
```python
401
import time
402
from pygls.exceptions import JsonRpcException
403
404
class RobustClient(JsonRPCClient):
405
def __init__(self, max_retries: int = 3):
406
super().__init__()
407
self.max_retries = max_retries
408
self.connected = False
409
410
async def connect_with_retry(self, host: str, port: int):
411
"""Connect with automatic retry logic."""
412
for attempt in range(self.max_retries):
413
try:
414
self.start_tcp(host, port)
415
await self.initialize_server()
416
self.connected = True
417
print(f"Connected to server on attempt {attempt + 1}")
418
return
419
420
except Exception as e:
421
print(f"Connection attempt {attempt + 1} failed: {e}")
422
if attempt < self.max_retries - 1:
423
await asyncio.sleep(2 ** attempt) # Exponential backoff
424
425
raise ConnectionError("Failed to connect after all retry attempts")
426
427
async def safe_request(self, method: str, params: Any, timeout: float = 10.0):
428
"""Send request with error handling and timeout."""
429
if not self.connected:
430
raise ConnectionError("Not connected to server")
431
432
try:
433
result = await asyncio.wait_for(
434
self.protocol.send_request(method, params),
435
timeout=timeout
436
)
437
return result
438
439
except asyncio.TimeoutError:
440
print(f"Request {method} timed out after {timeout}s")
441
return None
442
443
except JsonRpcException as e:
444
print(f"LSP error in {method}: {e}")
445
return None
446
447
except Exception as e:
448
print(f"Unexpected error in {method}: {e}")
449
return None
450
451
def connection_lost(self, exc):
452
"""Handle connection loss."""
453
self.connected = False
454
print(f"Connection lost: {exc}")
455
456
# Trigger reconnection logic if needed
457
asyncio.create_task(self.reconnect())
458
459
async def reconnect(self):
460
"""Attempt to reconnect to server."""
461
print("Attempting to reconnect...")
462
try:
463
await self.connect_with_retry("localhost", 8080)
464
except ConnectionError:
465
print("Reconnection failed")
466
```