1010import tomllib
1111from collections .abc import Awaitable , Callable
1212from pathlib import Path
13- from typing import Any
13+ from typing import TYPE_CHECKING , Any
1414
15- from mcp import ClientSession , StdioServerParameters
16- from mcp .client .stdio import stdio_client
17- from mcp .client .streamable_http import streamablehttp_client
15+ if TYPE_CHECKING :
16+ from mcp import ClientSession
1817
1918
2019EXPECTED_TOOLS : dict [str , set [str ]] = {
@@ -40,14 +39,11 @@ class ProbeFailure(Exception):
4039
4140
4241def _load_servers (root : Path , codex_home : Path ) -> dict [str , dict [str , Any ]]:
43- repo_path = root / "plugins/rldyour-mcps/.mcp.json"
42+ repo_servers = _load_repo_servers ( root )
4443 config_path = codex_home / "config.toml"
45- if not repo_path .is_file ():
46- raise ProbeFailure (f"missing { repo_path } " )
4744 if not config_path .is_file ():
4845 raise ProbeFailure (f"missing { config_path } " )
4946
50- repo_servers = json .loads (repo_path .read_text (encoding = "utf-8" ))["mcpServers" ]
5147 config_servers = tomllib .loads (config_path .read_text (encoding = "utf-8" )).get ("mcp_servers" , {})
5248 if set (repo_servers ) != set (config_servers ):
5349 raise ProbeFailure (
@@ -57,6 +53,90 @@ def _load_servers(root: Path, codex_home: Path) -> dict[str, dict[str, Any]]:
5753 return {name : dict (config_servers [name ]) for name in sorted (repo_servers )}
5854
5955
56+ def _load_repo_servers (root : Path ) -> dict [str , dict [str , Any ]]:
57+ repo_path = root / "plugins/rldyour-mcps/.mcp.json"
58+ if not repo_path .is_file ():
59+ raise ProbeFailure (f"missing { repo_path } " )
60+ try :
61+ payload = json .loads (repo_path .read_text (encoding = "utf-8" ))
62+ except json .JSONDecodeError as exc :
63+ raise ProbeFailure (f"{ repo_path } : invalid JSON: { exc } " ) from exc
64+ servers = payload .get ("mcpServers" )
65+ if not isinstance (servers , dict ):
66+ raise ProbeFailure (f"{ repo_path } : mcpServers must be an object" )
67+ return {str (name ): dict (spec ) for name , spec in sorted (servers .items ())}
68+
69+
70+ def _static_record (name : str , spec : dict [str , Any ]) -> tuple [dict [str , Any ], str | None ]:
71+ has_command = "command" in spec
72+ has_url = "url" in spec
73+ error : str | None = None
74+ if has_command == has_url :
75+ error = "must define exactly one of command or url"
76+ elif has_command and not isinstance (spec .get ("command" ), str ):
77+ error = "command must be a string"
78+ elif has_url and not isinstance (spec .get ("url" ), str ):
79+ error = "url must be a string"
80+ elif "args" in spec and not isinstance (spec .get ("args" ), list ):
81+ error = "args must be an array"
82+ elif "env" in spec and not isinstance (spec .get ("env" ), dict ):
83+ error = "env must be an object"
84+ elif "env_vars" in spec and not isinstance (spec .get ("env_vars" ), list ):
85+ error = "env_vars must be an array"
86+
87+ record : dict [str , Any ] = {
88+ "server" : name ,
89+ "status" : "static" if error is None else "fail" ,
90+ "transport" : "http" if has_url else "stdio" if has_command else "invalid" ,
91+ "expected_tools" : sorted (EXPECTED_TOOLS .get (name , set ())),
92+ }
93+ if has_command :
94+ record ["command" ] = spec .get ("command" )
95+ record ["args" ] = [str (arg ) for arg in spec .get ("args" ) or []]
96+ if has_url :
97+ record ["url" ] = spec .get ("url" )
98+ if "env" in spec :
99+ record ["env_keys" ] = sorted (str (key ) for key in (spec .get ("env" ) or {}))
100+ if "env_vars" in spec :
101+ record ["env_vars" ] = [str (key ) for key in spec .get ("env_vars" ) or []]
102+ if error :
103+ record ["error" ] = error
104+ return record , error
105+
106+
107+ def _main_static (args : argparse .Namespace ) -> int :
108+ root = args .root .resolve ()
109+ servers = _load_repo_servers (root )
110+ selected = set (args .server or servers )
111+ skipped = set (args .skip_server or [])
112+ records : list [dict [str , Any ]] = []
113+ errors : list [str ] = []
114+
115+ for name , spec in servers .items ():
116+ if name not in selected or name in skipped :
117+ continue
118+ record , error = _static_record (name , spec )
119+ records .append (record )
120+ if error :
121+ errors .append (f"{ name } : { error } " )
122+
123+ if args .json :
124+ print (json .dumps ({"mode" : "static" , "count" : len (records ), "results" : records , "errors" : errors }, indent = 2 ))
125+ else :
126+ print ("rldyour MCP capability smoke" )
127+ print (f"root: { root } " )
128+ print ("mode: static" )
129+ for record in records :
130+ prefix = "fail" if record ["status" ] == "fail" else "ok"
131+ detail = record .get ("error" ) or f"{ record ['transport' ]} config parsed"
132+ print (f"{ prefix :<7} { record ['server' ]} : { detail } " )
133+ if errors :
134+ print ("\n " .join (errors ), file = sys .stderr )
135+ else :
136+ print ("MCP static capability smoke passed." )
137+ return 1 if errors else 0
138+
139+
60140def _merged_env (spec : dict [str , Any ]) -> tuple [dict [str , str ], list [str ]]:
61141 env = dict (os .environ )
62142 missing : list [str ] = []
@@ -77,7 +157,7 @@ def _content_len(result: Any) -> int:
77157 return len (content ) if isinstance (content , list ) else 0
78158
79159
80- async def _safe_call (name : str , session : ClientSession , missing_env : list [str ]) -> str | None :
160+ async def _safe_call (name : str , session : " ClientSession" , missing_env : list [str ]) -> str | None :
81161 if name == "serena" :
82162 result = await session .call_tool ("list_memories" , {})
83163 if result .isError :
@@ -172,8 +252,11 @@ async def _safe_call(name: str, session: ClientSession, missing_env: list[str])
172252
173253async def _stdio_session (
174254 spec : dict [str , Any ],
175- body : Callable [[ClientSession , list [str ]], Awaitable [None ]],
255+ body : Callable [[" ClientSession" , list [str ]], Awaitable [None ]],
176256) -> None :
257+ from mcp import ClientSession , StdioServerParameters
258+ from mcp .client .stdio import stdio_client
259+
177260 command = str (spec .get ("command" ) or "" )
178261 if not command :
179262 raise ProbeFailure ("missing command" )
@@ -200,8 +283,11 @@ async def _stdio_session(
200283
201284async def _http_session (
202285 spec : dict [str , Any ],
203- body : Callable [[ClientSession , list [str ]], Awaitable [None ]],
286+ body : Callable [[" ClientSession" , list [str ]], Awaitable [None ]],
204287) -> None :
288+ from mcp import ClientSession
289+ from mcp .client .streamable_http import streamablehttp_client
290+
205291 url = str (spec .get ("url" ) or "" )
206292 if not url :
207293 raise ProbeFailure ("missing url" )
@@ -335,6 +421,13 @@ def _parse_args() -> argparse.Namespace:
335421 parser = argparse .ArgumentParser (description = "Probe MCP initialize/list_tools/safe call_tool behavior." )
336422 parser .add_argument ("--root" , type = Path , default = Path .cwd (), help = "Repository root." )
337423 parser .add_argument ("--codex-home" , type = Path , default = Path (os .environ .get ("CODEX_HOME" , "~/.codex" )))
424+ parser .add_argument (
425+ "--mode" ,
426+ choices = ("static" , "local-launch" ),
427+ default = "local-launch" ,
428+ help = "static parses repo MCP config only; local-launch probes the installed Codex MCP runtime." ,
429+ )
430+ parser .add_argument ("--json" , action = "store_true" , help = "Emit JSON. Supported for --mode static." )
338431 parser .add_argument ("--server" , action = "append" , help = "Only probe this server. Repeatable." )
339432 parser .add_argument ("--skip-server" , action = "append" , help = "Skip this server. Repeatable." )
340433 parser .add_argument ("--list-only" , action = "store_true" , help = "Only initialize and list tools." )
@@ -353,6 +446,11 @@ def _parse_args() -> argparse.Namespace:
353446
354447def main () -> int :
355448 args = _parse_args ()
449+ if args .mode == "static" :
450+ return _main_static (args )
451+ if args .json :
452+ print ("--json is only supported with --mode static" , file = sys .stderr )
453+ return 2
356454 return asyncio .run (_main_async (args ))
357455
358456
0 commit comments