Setting the file. One moment.
Mcps · Migrate To Codex · openai/skills · Skills Docs
Back to the top of the page
7 KB scripts/migrate/ mcps.py
Python · 222 lines · 7 KB
15 from collections.abc import Mapping
16 from pathlib import Path
17
18 from migrate.common import (
19 CODEX_CONFIG_PATH ,
20 MigrationReportItem,
21 json_object,
22 json_string,
23 json_string_tuple,
24 path_exists_with_exact_case,
25 )
26 from migrate.settings import CLAUDE_MCP_JSON_RELATIVE
27 from utils.util import TomlValue
28
29
30 ENV_VAR_RE = re.compile( r " \A \$\{ ([ A-Za-z_ ][ A-Za-z0-9_ ] * )(?: :- [ ^} ] * ) ? \} \Z " )
31 BEARER_ENV_VAR_RE = re.compile(
32 r " \A Bearer \s + \$\{ ([ A-Za-z_ ][ A-Za-z0-9_ ] * )(?: :- [ ^} ] * ) ? \} \Z "
33 )
34
35
36 def mcp_server_toml_table (
37 server_name: str ,
38 server_config: Mapping[ str , object ],
39 enabled_servers: tuple[ str , ... ],
40 disabled_servers: frozenset[ str ],
41 ) -> dict[ str , TomlValue]:
42 table: dict[ str , TomlValue] = {}
43 enabled = mcp_enabled_state(server_config)
44 if enabled is False :
45 table[ "enabled" ] = False
46 elif enabled_servers and server_name not in enabled_servers:
47 table[ "enabled" ] = False
48 elif server_name in disabled_servers:
49 table[ "enabled" ] = False
50 if url := json_string(server_config.get( "url" )):
51 table[ "url" ] = url
52 if command := json_string(server_config.get( "command" )):
53 table[ "command" ] = command
54 if args := json_string_tuple(server_config.get( "args" )):
55 table[ "args" ] = list (args)
56 if "headers" in server_config:
57 append_header_config(table, json_object(server_config[ "headers" ]))
58 if "env" in server_config:
59 append_env_config(table, json_object(server_config[ "env" ]))
60 return table
61
62
63 def mcp_report_items (
64 mcp_servers: tuple[tuple[ str , dict[ str , object ]], ... ],
65 ) -> list[MigrationReportItem]:
66 report_items = [
67 MigrationReportItem(
68 "rewritten" ,
69 CODEX_CONFIG_PATH ,
70 f "Converted { len (mcp_servers) } MCP server entries." ,
71 )
72 ]
73 for server_name, server_config in mcp_servers:
74 notes = mcp_manual_notes(server_name, server_config)
75 if notes:
76 report_items.append(
77 MigrationReportItem(
78 "manual_fix_required" ,
79 CODEX_CONFIG_PATH ,
80 f "MCP server ` { server_name } ` needs review: { ' ' .join(notes) } " ,
81 )
82 )
83 return report_items
84
85
86 def append_header_config (
87 table: dict[ str , TomlValue],
88 headers: Mapping[ str , object ],
89 ) -> None :
90 static_headers: dict[ str , str ] = {}
91 env_headers: dict[ str , str ] = {}
92
93 for key, value in headers.items():
94 header_value = str (value)
95 bearer_match = BEARER_ENV_VAR_RE .match(header_value)
96 if key.lower() == "authorization" and bearer_match:
97 table[ "bearer_token_env_var" ] = bearer_match.group( 1 )
98 continue
99
100 env_match = ENV_VAR_RE .match(header_value)
101 if env_match:
102 env_headers[key] = env_match.group( 1 )
103 continue
104
105 static_headers[key] = header_value
106
107 if static_headers:
108 table[ "http_headers" ] = static_headers
109 if env_headers:
110 table[ "env_http_headers" ] = env_headers
111
112
113 def append_env_config (
114 table: dict[ str , TomlValue],
115 env: Mapping[ str , object ],
116 ) -> None :
117 static_env: dict[ str , str ] = {}
118 env_vars: list[ str ] = []
119
120 for key, value in env.items():
121 env_value = str (value)
122 env_match = ENV_VAR_RE .match(env_value)
123 if env_match and env_match.group( 1 ) == key:
124 env_vars.append(key)
125 continue
126
127 static_env[key] = env_value
128
129 if env_vars:
130 table[ "env_vars" ] = env_vars
131 if static_env:
132 table[ "env" ] = static_env
133
134
135 def mcp_manual_notes (
136 server_name: str ,
137 server_config: Mapping[ str , object ],
138 ) -> tuple[ str , ... ]:
139 notes: list[ str ] = []
140 source_type = json_string(server_config.get( "type" ))
141 if source_type and source_type not in { "http" , "stdio" }:
142 notes.append(
143 f "Claude MCP `type: { source_type } ` was not written to Codex config; verify that the generated `url` or `command` is a supported Codex transport."
144 )
145 unsupported_fields = unsupported_mcp_server_fields(server_config)
146 if unsupported_fields:
147 notes.append(
148 "Review unsupported Claude MCP fields manually: "
149 + ", " .join( f "` { field_name } `" for field_name in unsupported_fields)
150 + "."
151 )
152 return tuple (notes)
153
154
155 def mcp_enabled_state (server_config: Mapping[ str , object ]) -> bool | None :
156 if server_config.get( "enabled" ) is False :
157 return False
158 if server_config.get( "disabled" ) is True :
159 return False
160 return None
161
162
163 def unsupported_mcp_server_fields (
164 server_config: Mapping[ str , object ],
165 ) -> tuple[ str , ... ]:
166 supported = {
167 "args" ,
168 "command" ,
169 "disabled" ,
170 "enabled" ,
171 "env" ,
172 "headers" ,
173 "name" ,
174 "scope" ,
175 "type" ,
176 "url" ,
177 }
178 return tuple ( sorted (key for key in server_config if key not in supported))
179
180
181 def read_claude_mcp_servers (source_root: Path) -> tuple[tuple[ str , dict[ str , object ]], ... ]:
182 servers: list[tuple[ str , dict[ str , object ]]] = []
183 for relative_path in CLAUDE_MCP_JSON_RELATIVE :
184 source_file = source_root / relative_path
185 if not path_exists_with_exact_case(source_file):
186 continue
187 mcp_config = json_object(json.loads(source_file.read_text()))
188 for server_name, server_config in json_object(mcp_config.get( "mcpServers" )).items():
189 servers.append((server_name, json_object(server_config)))
190 return tuple (servers)
191
192
193 def validate_mcp_commands (config: dict[ str , object ]) -> list[MigrationReportItem]:
194 mcp_servers = config.get( "mcp_servers" )
195 if not isinstance (mcp_servers, dict ):
196 return []
197
198 report_items: list[MigrationReportItem] = []
199 for server_name, server_config in sorted (mcp_servers.items()):
200 if not isinstance (server_config, dict ):
201 continue
202 command = server_config.get( "command" )
203 if not command:
204 continue
205 command_text = str (command)
206 if shutil.which(command_text):
207 report_items.append(
208 MigrationReportItem(
209 "ok" ,
210 CODEX_CONFIG_PATH ,
211 f "MCP server ` { server_name } ` command ` { command_text } ` is on PATH." ,
212 )
213 )
214 else :
215 report_items.append(
216 MigrationReportItem(
217 "warning" ,
218 CODEX_CONFIG_PATH ,
219 f "MCP server ` { server_name } ` command ` { command_text } ` was not found on PATH." ,
220 )
221 )
222 return report_items