|
8 | 8 |
|
9 | 9 | import platform |
10 | 10 | import json |
| 11 | +import tomllib # Python 3.11+ built-in |
| 12 | +import tomli_w # TOML writing |
11 | 13 | from pathlib import Path |
12 | | -from typing import Optional, Dict, Any |
| 14 | +from typing import Optional, Dict, Any, TextIO |
13 | 15 | import logging |
14 | 16 |
|
15 | 17 | from .host_management import MCPHostStrategy, register_host_strategy |
@@ -607,3 +609,157 @@ def write_configuration(self, config: HostConfiguration, no_backup: bool = False |
607 | 609 | except Exception as e: |
608 | 610 | logger.error(f"Failed to write Gemini configuration: {e}") |
609 | 611 | 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