code-analysis-client
Async Python client for the code-analysis server. It wraps mcp-proxy-adapter’s JsonRpcClient, so you get the adapter’s built-in methods (queue, transfer, help, health, …) plus thin helpers to run any registered server command.
Install
pip install code-analysis-client
Usage
import asyncio
from code_analysis_client import CodeAnalysisAsyncClient
async def main() -> None:
client = CodeAnalysisAsyncClient(
protocol="https",
host="127.0.0.1",
port=15001,
cert="/path/client.crt",
key="/path/client.key",
ca="/path/ca.crt",
timeout=120.0,
)
async with client:
h = await client.rpc.help()
r = await client.call("list_projects", {"include_deleted": False})
print(h, r)
asyncio.run(main())
Build client settings from the same JSON shape as the pipeline adapter settings (host, port, protocol, optional ssl with cert / key / ca or *_path aliases), or from a full server config.json object.
from code_analysis_client import CodeAnalysisAsyncClient
client = CodeAnalysisAsyncClient.from_server_config(config_dict, timeout=60.0)
Every call is the adapter's standard call
Every entry point — call, call_validated, client.commands.<name>, and the
file_sessions / universal_files / worktrees / check_chains facades built on
call_validated — goes through one core that uses mcp-proxy-adapter's
JsonRpcClient.execute_command_unified: the command runs in a WebSocket command
session and the call returns when the terminal event arrives (no polling while the
socket is up). If /ws is unreachable the adapter falls back to RPC polling of
queue_get_job_status and logs one WARNING WS_UNREACHABLE_RPC_POLLING_FALLBACK;
the result is identical. The client has no polling loop of its own.
# Queued or not, this waits and returns the command's own result envelope.
out = await client.call("some_long_running_command", {...})
Failures raise instead of returning an error envelope:
CommandFailedError— the command ran but its inner result is{"success": false}(immediate and queued commands alike; a failing result is never returned as a dict). Carries.command,.job_id,.error(the server's error object, e.g..error["code"]) and.result(the inner envelope).JobFailedError— the job failed/stopped/cancelled. Carries.job_id,.statusand.error(always a structured error dict).JobTimeoutError— only when a positivetimeoutelapses; the call detaches and the job keeps running server-side (websocket.on_timeoutisdetach).ClientValidationError— the server refused the parameters (JSON-RPC-32602), answered at the top level or carried by the failed job of an in-process command (same textJSON-RPC error: <message> (code: -32602),codeVALIDATION_ERROR);detailscarries the server's errordata(e.g. itsviolationpayload). A command the server refuses before starting it with its own error code (for exampleVALIDATION_ERRORfor an unknownproject_id) raises it with that exactcodeand the server'sdataasdetails.ValueError— a negativetimeoutor a non-positivepoll_interval, before anything is sent.
Optional keyword args on call / call_validated (and their call_unified*
counterparts):
timeout— total wait in seconds.None(default) uses the client configwebsocket.job_wait_timeout, shipped as0= no time limit, so a long queued command is never cut off. A value above 0 raisesJobTimeoutErrorand detaches.poll_interval— safety-net status-poll cadence in seconds.None(default) useswebsocket.terminal_poll_interval(2 s); a value overrides it per call.status_hook— sync or async callable invoked with each adapter WebSocket event (at least the accepted event first and the terminal completed event last). An exception raised by the hook propagates.
call / call_validated also take auto_poll (default True). Pass
auto_poll=False to detach as soon as the job id is known: the call returns a
QueuedJob handle immediately — for every command, including ones that finish at
once — and timeout, poll_interval and status_hook are ignored.
from code_analysis_client import QueuedJob
job = await client.call("some_long_running_command", {...}, auto_poll=False)
assert isinstance(job, QueuedJob)
# do other work here, then block on it whenever you're ready
result = await job.wait() # same result / errors as the default path
QueuedJob exposes .job_id, .envelope (the adapter's acceptance envelope) and two
async methods: .wait(timeout=None, poll_interval=None, status_hook=None) — reads the
job once and, if it is still running, uses the adapter's WebSocket wait for the job id
(public API wait_for_job_via_websocket), then returns/raises exactly like the default
path; there is no RPC-polling fallback for an existing job id (a WebSocket failure
propagates) — and .status() — a single queue_get_job_status fetch.
Explicit adapter timeout configuration
The client hands JsonRpcClient(ws_config=...) the explicit defaults
code_analysis_client.DEFAULT_CLIENT_ADAPTER_CONFIG (a ws_config= keyword you pass is
merged over them key by key), so none of the adapter's own 60 s / cancel defaults is
effective:
websocket key |
value | meaning |
|---|---|---|
job_wait_timeout |
0 | no total bound on a call's wait |
on_timeout |
detach |
a wait timeout leaves the job running |
receive_timeout / receive_timeout_buffer |
60 / 5 | one receive cap (never limits the total wait) |
heartbeat_interval |
30 | WebSocket heartbeat |
terminal_poll_interval |
2 | safety-net poll cadence |
close_timeout |
5 | socket close bound |
subscribe_settle_delay / cancel_grace_seconds |
0.5 / 0.5 | settle delay, cancel grace |
job_push: on_caller_loss detach (a job survives a lost client), poll_interval 0.1.
The server ships and applies the same values.
call_unified / call_unified_validated are kept as deprecated aliases of
call / call_validated for backward compatibility and emit
DeprecationWarning on every call. expect_queue remains accepted-and-ignored
(documented no-op). auto_poll is the one canonical switch and is forwarded
straight through to call / call_validated — auto_poll=False on an alias
returns a QueuedJob the same way it does on the non-deprecated method.
Prefer call / call_validated directly.
Validation using the server schema
The authoritative input schema is whatever the running server returns from help with cmdname set to the command. The client calls that, optionally caches the result, performs the same shallow checks as the server’s BaseMCPCommand (types, required, enum, additionalProperties), then runs the command.
async with CodeAnalysisAsyncClient(host="127.0.0.1", port=15001) as client:
# Explicit
out = await client.call_validated(
"list_projects",
{"include_deleted": False},
)
# Dynamic wrapper: same as call_validated("list_projects", {...})
out = await client.commands.list_projects(include_deleted=False)
# After server reload
client.clear_command_schema_cache()
Pass refresh_schema=True on a single call to bypass the in-memory schema cache.
High-level facades (aligned with live server registry)
The client does not wrap CST commands (cst_load_file, …) or legacy file I/O
(universal_file_read, read_project_text_file, …). Those commands are removed
from the server registry. Use the facades below or generic call / commands.*.
| Facade | Property | Server commands |
|---|---|---|
| Client DB sessions + transfer | client.file_sessions |
session_*, subordinate_session_*, project_file_transfer_*, project_file_advisory_lock_batch |
| Universal file preview | client.universal_files |
universal_file_preview (read-only) |
| Any registered command | client.call / client.commands.<name> |
schema from live help() |
Canonical command lists: code_analysis_client.server_api — exported as
FILE_SESSION_COMMANDS, FILE_SESSION_FACADE_METHODS, CLIENT_FACADE_COMMANDS,
REMOVED_COMMANDS.
Scope boundary: this client manipulates files only as whole units — transfer, locks, sessions, and structured read-only preview — and analyzes them. Content editing (open/edit/write/close draft sessions) is not served by this project's code-analysis server; use the ai-editor client for that.
Sync checks (in-process registry):
pytest tests/test_client_server_api_sync.py tests/test_code_analysis_client.py -k session
Package version is in the root pyproject.toml; before a client wheel build run
python scripts/sync_code_analysis_client_version.py (./build.sh <VERSION> does it as its bump step).
Examples (this repository)
Runnable scripts live under client/examples/. Long-form “man page” style
documentation is embedded in the module docstrings of those Python files
(see client/examples/README.md for how to read them).
| Script | Purpose |
|---|---|
run_all_examples.py |
Full API tour + runs all live sibling scripts |
ex_minimal_validated.py |
Smallest validated RPC example |
ex_universal_files.py |
UniversalFileClient.preview (read-only structured preview) |
ex_session_view_subordinates.py |
session_view and subordinate CRUD |
ex_file_sessions.py |
Sessions, locks, transfer roundtrip |
ex_config_only.py |
Parse config.json without TCP |
casmgr --config config.json start
python client/examples/run_all_examples.py
Development
From the repository root:
pip install -e ./client
pytest tests/test_code_analysis_client.py
Releasing to PyPI (version = root code-analysis project)
The client wheel version is read from client/code_analysis_client/version.txt.
That file must match [project].version in the repository root
pyproject.toml. The client is published only by the repository release script,
together with the server image (same version, already-published versions fail):
./build.sh <VERSION>
1 maintainer
Metadata
1 maintainer
Release files for code-analysis-client 1.6.245
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| code_analysis_client-1.6.245.tar.gz | 44.1 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| code_analysis_client-1.6.245-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 92.2 kB
Release files / code_analysis_client-1.6.245.tar.gz
| Download URL | code_analysis_client-1.6.245.tar.gz |
|---|---|
| Size | 44.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
f6e50ed28507ac439f574f3a09a81b264ca1436a61faa44ac9c25a683dde5404
|
|
BLAKE2b-256 checksum How to use checksums |
faec759caed46e6e1a83cebeb6a777da66356039f33d03f50ad1e132af8d5373
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.14.4
|
Release files / code_analysis_client-1.6.245-py3-none-any.whl
| Download URL | code_analysis_client-1.6.245-py3-none-any.whl |
|---|---|
| Size | 48.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
97e1faedc5e75127d70cf9e70e69f7873fe112ceb6e949da80c1cf3f53fce492
|
|
BLAKE2b-256 checksum How to use checksums |
443bd361788f55c028f74a3354f5408d0f715711c254d3a524c6990926721277
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.14.4
|
Security for code-analysis-client 1.6.245
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
