Skip to content

Commit cac2301

Browse files
LittleCoinCoinLittleCoinCoin
authored andcommitted
feat(codex): implement CodexHostStrategy with TOML support
Implement complete CodexHostStrategy for Codex IDE configuration: Strategy Features: - Config path: ~/.codex/config.toml - Config key: 'mcp_servers' (underscore, not camelCase) - Format: TOML with nested [mcp_servers.<name>] tables - Host detection: Checks for ~/.codex directory - Validation: Supports both STDIO and HTTP servers TOML Handling: - Read: Uses Python 3.12+ built-in tomllib - Write: Uses tomli_w library - Preserves [features] section during read/write - Preserves other top-level TOML keys - Handles nested [mcp_servers.<name>.env] tables Backup Integration: - Uses atomic_write_with_serializer() for TOML - Creates backups with pattern: config.toml.codex.{timestamp} - Supports no_backup parameter - Integrates with MCPHostConfigBackupManager Methods: - get_config_path(), get_config_key(), is_host_available() - validate_server_config(), read_configuration(), write_configuration() - _flatten_toml_server(), _to_toml_server() (TOML conversion helpers) Registered via @register_host_strategy(MCPHostType.CODEX) decorator.
1 parent ed86ddf commit cac2301

File tree

1 file changed

+157
-1
lines changed

1 file changed

+157
-1
lines changed

hatch/mcp_host_config/strategies.py

Lines changed: 157 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,10 @@
88

99
import platform
1010
import json
11+
import tomllib # Python 3.11+ built-in
12+
import tomli_w # TOML writing
1113
from pathlib import Path
12-
from typing import Optional, Dict, Any
14+
from typing import Optional, Dict, Any, TextIO
1315
import logging
1416

1517
from .host_management import MCPHostStrategy, register_host_strategy
@@ -607,3 +609,157 @@ def write_configuration(self, config: HostConfiguration, no_backup: bool = False
607609
except Exception as e:
608610
logger.error(f"Failed to write Gemini configuration: {e}")
609611
return False
612+
613+
614+
@register_host_strategy(MCPHostType.CODEX)
615+
class CodexHostStrategy(MCPHostStrategy):
616+
"""Configuration strategy for Codex IDE with TOML support.
617+
618+
Codex uses TOML configuration at ~/.codex/config.toml with a unique
619+
structure using [mcp_servers.<server-name>] tables.
620+
"""
621+
622+
def __init__(self):
623+
self.config_format = "toml"
624+
self._preserved_features = {} # Preserve [features] section
625+
626+
def get_config_path(self) -> Optional[Path]:
627+
"""Get Codex configuration path."""
628+
return Path.home() / ".codex" / "config.toml"
629+
630+
def get_config_key(self) -> str:
631+
"""Codex uses 'mcp_servers' key (note: underscore, not camelCase)."""
632+
return "mcp_servers"
633+
634+
def is_host_available(self) -> bool:
635+
"""Check if Codex is available by checking for config directory."""
636+
codex_dir = Path.home() / ".codex"
637+
return codex_dir.exists()
638+
639+
def validate_server_config(self, server_config: MCPServerConfig) -> bool:
640+
"""Codex validation - supports both STDIO and HTTP servers."""
641+
return server_config.command is not None or server_config.url is not None
642+
643+
def read_configuration(self) -> HostConfiguration:
644+
"""Read Codex TOML configuration file."""
645+
config_path = self.get_config_path()
646+
if not config_path or not config_path.exists():
647+
return HostConfiguration(servers={})
648+
649+
try:
650+
with open(config_path, 'rb') as f:
651+
toml_data = tomllib.load(f)
652+
653+
# Preserve [features] section for later write
654+
self._preserved_features = toml_data.get('features', {})
655+
656+
# Extract MCP servers from [mcp_servers.*] tables
657+
mcp_servers = toml_data.get(self.get_config_key(), {})
658+
659+
servers = {}
660+
for name, server_data in mcp_servers.items():
661+
try:
662+
# Flatten nested env section if present
663+
flat_data = self._flatten_toml_server(server_data)
664+
servers[name] = MCPServerConfig(**flat_data)
665+
except Exception as e:
666+
logger.warning(f"Invalid server config for {name}: {e}")
667+
continue
668+
669+
return HostConfiguration(servers=servers)
670+
671+
except Exception as e:
672+
logger.error(f"Failed to read Codex configuration: {e}")
673+
return HostConfiguration(servers={})
674+
675+
def write_configuration(self, config: HostConfiguration, no_backup: bool = False) -> bool:
676+
"""Write Codex TOML configuration file with backup support."""
677+
config_path = self.get_config_path()
678+
if not config_path:
679+
return False
680+
681+
try:
682+
config_path.parent.mkdir(parents=True, exist_ok=True)
683+
684+
# Read existing configuration to preserve non-MCP settings
685+
existing_data = {}
686+
if config_path.exists():
687+
try:
688+
with open(config_path, 'rb') as f:
689+
existing_data = tomllib.load(f)
690+
except Exception:
691+
pass
692+
693+
# Preserve [features] section
694+
if 'features' in existing_data:
695+
self._preserved_features = existing_data['features']
696+
697+
# Convert servers to TOML structure
698+
servers_data = {}
699+
for name, server_config in config.servers.items():
700+
servers_data[name] = self._to_toml_server(server_config)
701+
702+
# Build final TOML structure
703+
final_data = {}
704+
705+
# Preserve [features] at top
706+
if self._preserved_features:
707+
final_data['features'] = self._preserved_features
708+
709+
# Add MCP servers
710+
final_data[self.get_config_key()] = servers_data
711+
712+
# Preserve other top-level keys
713+
for key, value in existing_data.items():
714+
if key not in ('features', self.get_config_key()):
715+
final_data[key] = value
716+
717+
# Use atomic write with TOML serializer
718+
backup_manager = MCPHostConfigBackupManager()
719+
atomic_ops = AtomicFileOperations()
720+
721+
def toml_serializer(data: Any, f: TextIO) -> None:
722+
# tomli_w.dumps returns a string, write it to the file
723+
toml_str = tomli_w.dumps(data)
724+
f.write(toml_str)
725+
726+
atomic_ops.atomic_write_with_serializer(
727+
file_path=config_path,
728+
data=final_data,
729+
serializer=toml_serializer,
730+
backup_manager=backup_manager,
731+
hostname="codex",
732+
skip_backup=no_backup
733+
)
734+
735+
return True
736+
737+
except Exception as e:
738+
logger.error(f"Failed to write Codex configuration: {e}")
739+
return False
740+
741+
def _flatten_toml_server(self, server_data: Dict[str, Any]) -> Dict[str, Any]:
742+
"""Flatten nested TOML server structure to flat dict.
743+
744+
TOML structure:
745+
[mcp_servers.name]
746+
command = "npx"
747+
args = ["-y", "package"]
748+
[mcp_servers.name.env]
749+
VAR = "value"
750+
751+
Becomes:
752+
{"command": "npx", "args": [...], "env": {"VAR": "value"}}
753+
"""
754+
# TOML already parses nested tables into nested dicts
755+
# So [mcp_servers.name.env] becomes {"env": {...}}
756+
return dict(server_data)
757+
758+
def _to_toml_server(self, server_config: MCPServerConfig) -> Dict[str, Any]:
759+
"""Convert MCPServerConfig to TOML-compatible dict structure."""
760+
data = server_config.model_dump(exclude_unset=True)
761+
762+
# Remove 'name' field as it's the table key in TOML
763+
data.pop('name', None)
764+
765+
return data

0 commit comments

Comments
 (0)