Skip to content

agentcore_cli.services.config

agentcore_cli.services.config

Configuration management for AgentCore Platform CLI.

This module provides a centralized configuration management system for the AgentCore Platform CLI, including local file storage and cloud synchronization.

ConfigManager

Centralized configuration manager for AgentCore Platform CLI.

This class is responsible for loading, saving, and managing the configuration for the AgentCore Platform CLI. It provides a clean interface for accessing and modifying configuration settings.

The configuration is stored in a local JSON file and can be synchronized with AWS Parameter Store when cloud sync is enabled.

Attributes:

Name Type Description
config AgentCoreConfig

The current configuration.

config_dir str

Directory where configuration is stored.

config_file str

Path to the configuration file.

Source code in agentcore_cli/services/config.py
Python
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
class ConfigManager:
    """Centralized configuration manager for AgentCore Platform CLI.

    This class is responsible for loading, saving, and managing the configuration
    for the AgentCore Platform CLI. It provides a clean interface for accessing
    and modifying configuration settings.

    The configuration is stored in a local JSON file and can be synchronized with
    AWS Parameter Store when cloud sync is enabled.

    Attributes:
        config (AgentCoreConfig): The current configuration.
        config_dir (str): Directory where configuration is stored.
        config_file (str): Path to the configuration file.
    """

    def __init__(self) -> None:
        """Initialize the configuration manager."""
        self.config = AgentCoreConfig()
        self.config_dir = os.path.join(os.getcwd(), ".agentcore")
        self.config_file = os.path.join(self.config_dir, "config.json")
        self._load_config()

    def _load_config(self) -> None:
        """Load configuration from local file.

        If the file doesn't exist or can't be parsed, a default configuration
        is used.
        """
        try:
            # Create config directory if it doesn't exist
            os.makedirs(self.config_dir, exist_ok=True)

            # Check if config file exists
            if not os.path.exists(self.config_file):
                logger.debug(f"Configuration file not found at {self.config_file}")
                self._create_default_config()
                self.save_config()
                return

            # Load the config file
            with open(self.config_file, "r") as f:
                config_data = json.load(f)

            # Parse the config data
            self.config = AgentCoreConfig.model_validate(config_data)
            self.config.config_path = self.config_file

            logger.debug(f"Configuration loaded from {self.config_file}")

        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse configuration file: {str(e)}")
            self._create_default_config()
            self.save_config()
        except Exception as e:
            logger.error(f"Error loading configuration: {str(e)}")
            self._create_default_config()

    def _create_default_config(self) -> None:
        """Create a default configuration."""
        # Create default environment
        default_env = EnvironmentConfig(name="dev", region="us-west-2")
        # Set creation timestamp manually since it's in the validator
        default_env.created_at = datetime.now()

        # Create default global resources
        from agentcore_cli.models.config import GlobalResourceConfig

        global_resources = GlobalResourceConfig()

        # Create default config with minimal parameters
        self.config = AgentCoreConfig()

        # Set properties after creation
        self.config.current_environment = "dev"
        self.config.environments = {"dev": default_env}
        self.config.global_resources = global_resources
        self.config.config_path = self.config_file

        logger.debug("Created default configuration")

    def save_config(self) -> bool:
        """Save current configuration to file."""
        try:
            # Ensure the config directory exists
            os.makedirs(self.config_dir, exist_ok=True)

            # Convert to dict and save
            config_dict = self.config.model_dump(mode="json")

            # Remove config_path to avoid circular references
            if "config_path" in config_dict:
                del config_dict["config_path"]

            with open(self.config_file, "w") as f:
                json.dump(config_dict, f, indent=2, default=str)

            logger.debug(f"Configuration saved to {self.config_file}")

            # Perform auto-sync if enabled
            if (
                self.config.global_resources.sync_config
                and self.config.global_resources.sync_config.cloud_config_enabled
            ):
                self.sync_with_cloud(auto=True)

            return True
        except Exception as e:
            logger.error(f"Failed to save configuration: {str(e)}")
            return False

    def sync_with_cloud(self, auto: bool = False) -> CloudSyncResult:
        """Sync configuration with AWS Parameter Store.

        Args:
            auto: Whether this is an automatic sync.

        Returns:
            CloudSyncResult: Result of the sync operation.
        """
        from agentcore_cli.services.config_sync import ConfigSyncService

        # Only sync if cloud sync is enabled
        if not self.config.global_resources.sync_config.cloud_config_enabled:
            return CloudSyncResult(
                success=False,
                message="Cloud sync is not enabled",
                environment=self.config.current_environment,
                synced_items={},
                errors=["Cloud sync is not enabled. Enable it first with 'config sync --enable'."],
            )

        # Only auto-sync if auto-sync is enabled
        if auto and (not self.config.global_resources.sync_config.auto_sync_enabled):
            return CloudSyncResult(
                success=True,
                message="Auto-sync is disabled",
                environment=self.config.current_environment,
                synced_items={},
                errors=[],
            )

        # Get current environment region
        region = self.get_region()

        # Create sync service
        sync_service = ConfigSyncService(region=region, config=self.config)

        # Check if we should auto-sync
        if auto and not sync_service.should_auto_sync:
            return CloudSyncResult(
                success=True,
                message="Automatic sync not needed",
                environment=self.config.current_environment,
                synced_items={},
                errors=[],
            )

        # Push config to cloud
        return sync_service.push_config_to_cloud(self.config)

    def pull_from_cloud(self) -> bool:
        """Pull configuration from AWS Parameter Store.

        Returns:
            bool: True if successful, False otherwise.
        """
        from agentcore_cli.services.config_sync import ConfigSyncService

        # Only sync if cloud sync is enabled
        if (
            not self.config.global_resources.sync_config
            or not self.config.global_resources.sync_config.cloud_config_enabled
        ):
            logger.error("Cloud sync is not enabled")
            return False

        # Get current environment region
        region = self.get_region()

        # Create sync service
        sync_service = ConfigSyncService(region=region, config=self.config)

        # Pull config from cloud
        success, config_dict, errors = sync_service.pull_config_from_cloud()

        if not success:
            logger.error(f"Failed to pull configuration from cloud: {errors}")
            return False

        try:
            # Update config with cloud data
            self.config = AgentCoreConfig.model_validate(config_dict)
            self.config.config_path = self.config_file

            # Save the updated config
            self.save_config()

            return True
        except Exception as e:
            logger.error(f"Failed to update configuration with cloud data: {str(e)}")
            return False

    def get_cloud_sync_status(self, environment: str | None = None) -> SyncStatus:
        """Get the sync status between local and cloud configuration.

        Args:
            environment: Optional environment to check. If None, checks current environment.

        Returns:
            SyncStatus: Sync status between local and cloud.
        """
        from agentcore_cli.services.config_sync import ConfigSyncService

        env = environment or self.config.current_environment
        region = self.get_environment(env).region

        sync_service = ConfigSyncService(region=region, config=self.config)
        return sync_service.check_sync_status(self.config, env)

    def enable_cloud_sync(self, enable: bool = True) -> bool:
        """Enable or disable cloud configuration sync.

        Args:
            enable: Whether to enable or disable cloud sync.

        Returns:
            bool: True if successful, False otherwise.
        """
        from agentcore_cli.services.config_sync import ConfigSyncService

        region = self.get_region()
        sync_service = ConfigSyncService(region=region, config=self.config)

        result = sync_service.enable_cloud_sync(self.config, enable)

        if result:
            self.save_config()

        return result

    def enable_auto_sync(self, enable: bool = True) -> bool:
        """Enable or disable automatic configuration sync.

        Args:
            enable: Whether to enable or disable auto-sync.

        Returns:
            bool: True if successful, False otherwise.
        """
        from agentcore_cli.services.config_sync import ConfigSyncService

        region = self.get_region()
        sync_service = ConfigSyncService(region=region, config=self.config)

        result = sync_service.enable_auto_sync(self.config, enable)

        if result:
            self.save_config()

        return result

    def set_sync_interval(self, minutes: int) -> bool:
        """Set the sync interval for auto-sync.

        Args:
            minutes: Sync interval in minutes.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            if not self.config.global_resources.sync_config:
                self.config.global_resources.sync_config = SyncConfig()

            self.config.global_resources.sync_config.sync_interval_minutes = minutes
            self.save_config()

            return True
        except Exception as e:
            logger.error(f"Failed to set sync interval: {str(e)}")
            return False

    def get_region(self, environment: str | None = None) -> str:
        """Get the AWS region for an environment.

        Args:
            environment: Optional environment name. If None, uses current environment.

        Returns:
            str: AWS region.
        """
        env_name = environment or self.config.current_environment

        if env_name in self.config.environments:
            return self.config.environments[env_name].region
        else:
            # Default to us-west-2
            return "us-west-2"

    @property
    def current_environment(self) -> str:
        """Get the current environment name.

        Returns:
            str: Current environment name.
        """
        return self.config.current_environment

    def set_current_environment(self, env_name: str) -> bool:
        """Set the current environment.

        Args:
            env_name: Environment name.

        Returns:
            bool: True if successful, False otherwise.
        """
        # Check if environment exists
        if env_name not in self.config.environments:
            logger.error(f"Environment '{env_name}' does not exist")
            return False

        # Set current environment
        self.config.current_environment = env_name
        self.save_config()

        logger.debug(f"Current environment set to '{env_name}'")
        return True

    def get_environment(self, name: str | None = None) -> EnvironmentConfig:
        """Get an environment configuration.

        Args:
            name: Optional environment name. If None, uses current environment.

        Returns:
            EnvironmentConfig: Environment configuration.

        Raises:
            KeyError: If the environment does not exist.
        """
        env_name = name or self.config.current_environment

        if env_name not in self.config.environments:
            raise KeyError(f"Environment '{env_name}' does not exist")

        return self.config.environments[env_name]

    def add_environment(self, name: str, region: str) -> bool:
        """Add a new environment.

        Args:
            name: Environment name.
            region: AWS region.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Check if environment already exists
            if name in self.config.environments:
                logger.error(f"Environment '{name}' already exists")
                return False

            # Create environment
            env = EnvironmentConfig(name=name, region=region, created_at=datetime.now())

            # Add to config
            self.config.environments[name] = env

            # Save config
            self.save_config()

            logger.info(f"Environment '{name}' added")
            return True
        except Exception as e:
            logger.error(f"Failed to add environment: {str(e)}")
            return False

    def update_environment(self, name: str, **kwargs: Any) -> bool:
        """Update an environment.

        Args:
            name: Environment name.
            **kwargs: Environment properties to update.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Check if environment exists
            if name not in self.config.environments:
                logger.error(f"Environment '{name}' does not exist")
                return False

            # Get environment
            env = self.config.environments[name]

            # Update properties
            for key, value in kwargs.items():
                if hasattr(env, key):
                    setattr(env, key, value)

            # Update timestamp
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Environment '{name}' updated")
            return True
        except Exception as e:
            logger.error(f"Failed to update environment: {str(e)}")
            return False

    def delete_environment(self, name: str) -> bool:
        """Delete an environment.

        Args:
            name: Environment name.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Check if environment exists
            if name not in self.config.environments:
                logger.error(f"Environment '{name}' does not exist")
                return False

            # Check if it's the current environment
            if name == self.config.current_environment:
                logger.error(f"Cannot delete current environment '{name}'")
                return False

            # Delete environment
            del self.config.environments[name]

            # Save config
            self.save_config()

            logger.info(f"Environment '{name}' deleted")
            return True
        except Exception as e:
            logger.error(f"Failed to delete environment: {str(e)}")
            return False

    def get_agent_runtime(self, name: str, environment: str | None = None) -> AgentRuntime | None:
        """Get an agent runtime by name from the specified or current environment.

        Args:
            name: Agent runtime name.
            environment: Environment name. If None, uses current environment.

        Returns:
            Optional[AgentRuntime]: Agent runtime if found, None otherwise.
        """
        env_name = environment or self.config.current_environment
        try:
            env = self.get_environment(env_name)
            return env.agent_runtimes.get(name)
        except KeyError:
            logger.error(f"Environment '{env_name}' does not exist")
            return None

    def add_agent_runtime(self, name: str, runtime_config: AgentRuntime, environment: str | None = None) -> bool:
        """Add an agent runtime to the specified or current environment.

        Args:
            name: Agent runtime name.
            runtime_config: Agent runtime configuration.
            environment: Environment name. If None, uses current environment.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            env_name = environment or self.config.current_environment
            env = self.get_environment(env_name)

            # Ensure the runtime has the correct region
            runtime_config.region = env.region

            # Set or update the agent runtime in the environment
            env.agent_runtimes[name] = runtime_config

            # Update environment timestamp
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Agent runtime '{name}' added to environment '{env_name}'")
            return True
        except KeyError:
            logger.error(f"Environment '{env_name}' does not exist")
            return False
        except Exception as e:
            logger.error(f"Failed to add agent runtime: {str(e)}")
            return False

    def update_agent_runtime(self, name: str, environment: str | None = None, **kwargs: Any) -> bool:
        """Update an agent runtime in the specified or current environment.

        Args:
            name: Agent runtime name.
            environment: Environment name. If None, uses current environment.
            **kwargs: Runtime properties to update.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            env_name = environment or self.config.current_environment
            env = self.get_environment(env_name)

            # Check if agent runtime exists in this environment
            if name not in env.agent_runtimes:
                logger.error(f"Agent runtime '{name}' does not exist in environment '{env_name}'")
                return False

            # Get agent runtime
            runtime = env.agent_runtimes[name]

            # Update properties
            for key, value in kwargs.items():
                if hasattr(runtime, key):
                    setattr(runtime, key, value)

            # Update timestamp
            runtime.updated_at = datetime.now()
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Agent runtime '{name}' updated in environment '{env_name}'")
            return True
        except KeyError:
            logger.error(f"Environment '{env_name}' does not exist")
            return False
        except Exception as e:
            logger.error(f"Failed to update agent runtime: {str(e)}")
            return False

    def delete_agent_runtime(self, name: str, environment: str | None = None) -> bool:
        """Delete an agent runtime from the specified or current environment.

        Args:
            name: Agent runtime name.
            environment: Environment name. If None, uses current environment.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            env_name = environment or self.config.current_environment
            env = self.get_environment(env_name)

            # Check if agent runtime exists in this environment
            if name not in env.agent_runtimes:
                logger.error(f"Agent runtime '{name}' does not exist in environment '{env_name}'")
                return False

            # Delete agent runtime
            del env.agent_runtimes[name]

            # Update environment timestamp
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Agent runtime '{name}' deleted from environment '{env_name}'")
            return True
        except KeyError:
            logger.error(f"Environment '{env_name}' does not exist")
            return False
        except Exception as e:
            logger.error(f"Failed to delete agent runtime: {str(e)}")
            return False

    def add_ecr_repository(self, name: str, repo_config: ECRRepository) -> bool:
        """Add an ECR repository to global resources.

        Args:
            name: Repository name.
            repo_config: Repository configuration.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Set or update the ECR repository in global resources
            self.config.global_resources.ecr_repositories[name] = repo_config

            # Save config
            self.save_config()

            logger.info(f"ECR repository '{name}' added to global resources")
            return True
        except Exception as e:
            logger.error(f"Failed to add ECR repository: {str(e)}")
            return False

    def add_iam_role(self, name: str, role_config: IAMRoleConfig) -> bool:
        """Add an IAM role to global resources.

        Args:
            name: Role name.
            role_config: Role configuration.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Set or update the IAM role in global resources
            self.config.global_resources.iam_roles[name] = role_config

            # Save config
            self.save_config()

            logger.info(f"IAM role '{name}' added to global resources")
            return True
        except Exception as e:
            logger.error(f"Failed to add IAM role: {str(e)}")
            return False

    def add_cognito_config(self, env_name: str, cognito_config: CognitoConfig) -> bool:
        """Add a Cognito configuration to an environment.

        Args:
            env_name: Environment name.
            cognito_config: Cognito configuration.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Check if environment exists
            if env_name not in self.config.environments:
                logger.error(f"Environment '{env_name}' does not exist")
                return False

            # Get environment
            env = self.config.environments[env_name]

            # Set Cognito config
            env.cognito = cognito_config

            # Update timestamp
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Cognito configuration added to environment '{env_name}'")
            return True
        except Exception as e:
            logger.error(f"Failed to add Cognito configuration: {str(e)}")
            return False

    def set_default_agent_runtime(self, env_name: str, agent_name: str) -> bool:
        """Set the default agent runtime for an environment.

        Args:
            env_name: Environment name.
            agent_name: Agent runtime name.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Check if environment exists
            if env_name not in self.config.environments:
                logger.error(f"Environment '{env_name}' does not exist")
                return False

            # Get environment
            env = self.config.environments[env_name]

            # Check if agent runtime exists in this environment
            if agent_name not in env.agent_runtimes:
                logger.error(f"Agent runtime '{agent_name}' does not exist in environment '{env_name}'")
                return False

            # Set default agent runtime
            env.default_agent_runtime = agent_name

            # Update timestamp
            env.updated_at = datetime.now()

            # Save config
            self.save_config()

            logger.info(f"Default agent runtime set to '{agent_name}' in environment '{env_name}'")
            return True
        except Exception as e:
            logger.error(f"Failed to set default agent runtime: {str(e)}")
            return False

    def export_config(self, file_path: str) -> bool:
        """Export configuration to a file.

        Args:
            file_path: Path to export the configuration to.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Convert to dict
            config_dict = self.config.model_dump(mode="json")

            # Remove config_path to avoid circular references
            if "config_path" in config_dict:
                del config_dict["config_path"]

            # Write to file
            with open(file_path, "w") as f:
                json.dump(config_dict, f, indent=2, default=str)

            logger.info(f"Configuration exported to {file_path}")
            return True
        except Exception as e:
            logger.error(f"Failed to export configuration: {str(e)}")
            return False

    def import_config(self, file_path: str) -> bool:
        """Import configuration from a file.

        Args:
            file_path: Path to import the configuration from.

        Returns:
            bool: True if successful, False otherwise.
        """
        try:
            # Read file
            with open(file_path, "r") as f:
                config_data = json.load(f)

            # Parse config
            new_config = AgentCoreConfig.model_validate(config_data)

            # Keep the config path
            new_config.config_path = self.config_file

            # Update config
            self.config = new_config

            # Save config
            self.save_config()

            logger.info(f"Configuration imported from {file_path}")
            return True
        except Exception as e:
            logger.error(f"Failed to import configuration: {str(e)}")
            return False

current_environment property

Get the current environment name.

Returns:

Name Type Description
str str

Current environment name.

__init__()

Initialize the configuration manager.

Source code in agentcore_cli/services/config.py
Python
def __init__(self) -> None:
    """Initialize the configuration manager."""
    self.config = AgentCoreConfig()
    self.config_dir = os.path.join(os.getcwd(), ".agentcore")
    self.config_file = os.path.join(self.config_dir, "config.json")
    self._load_config()

add_agent_runtime(name, runtime_config, environment=None)

Add an agent runtime to the specified or current environment.

Parameters:

Name Type Description Default
name str

Agent runtime name.

required
runtime_config AgentRuntime

Agent runtime configuration.

required
environment str | None

Environment name. If None, uses current environment.

None

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def add_agent_runtime(self, name: str, runtime_config: AgentRuntime, environment: str | None = None) -> bool:
    """Add an agent runtime to the specified or current environment.

    Args:
        name: Agent runtime name.
        runtime_config: Agent runtime configuration.
        environment: Environment name. If None, uses current environment.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        env_name = environment or self.config.current_environment
        env = self.get_environment(env_name)

        # Ensure the runtime has the correct region
        runtime_config.region = env.region

        # Set or update the agent runtime in the environment
        env.agent_runtimes[name] = runtime_config

        # Update environment timestamp
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Agent runtime '{name}' added to environment '{env_name}'")
        return True
    except KeyError:
        logger.error(f"Environment '{env_name}' does not exist")
        return False
    except Exception as e:
        logger.error(f"Failed to add agent runtime: {str(e)}")
        return False

add_cognito_config(env_name, cognito_config)

Add a Cognito configuration to an environment.

Parameters:

Name Type Description Default
env_name str

Environment name.

required
cognito_config CognitoConfig

Cognito configuration.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def add_cognito_config(self, env_name: str, cognito_config: CognitoConfig) -> bool:
    """Add a Cognito configuration to an environment.

    Args:
        env_name: Environment name.
        cognito_config: Cognito configuration.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Check if environment exists
        if env_name not in self.config.environments:
            logger.error(f"Environment '{env_name}' does not exist")
            return False

        # Get environment
        env = self.config.environments[env_name]

        # Set Cognito config
        env.cognito = cognito_config

        # Update timestamp
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Cognito configuration added to environment '{env_name}'")
        return True
    except Exception as e:
        logger.error(f"Failed to add Cognito configuration: {str(e)}")
        return False

add_ecr_repository(name, repo_config)

Add an ECR repository to global resources.

Parameters:

Name Type Description Default
name str

Repository name.

required
repo_config ECRRepository

Repository configuration.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def add_ecr_repository(self, name: str, repo_config: ECRRepository) -> bool:
    """Add an ECR repository to global resources.

    Args:
        name: Repository name.
        repo_config: Repository configuration.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Set or update the ECR repository in global resources
        self.config.global_resources.ecr_repositories[name] = repo_config

        # Save config
        self.save_config()

        logger.info(f"ECR repository '{name}' added to global resources")
        return True
    except Exception as e:
        logger.error(f"Failed to add ECR repository: {str(e)}")
        return False

add_environment(name, region)

Add a new environment.

Parameters:

Name Type Description Default
name str

Environment name.

required
region str

AWS region.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def add_environment(self, name: str, region: str) -> bool:
    """Add a new environment.

    Args:
        name: Environment name.
        region: AWS region.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Check if environment already exists
        if name in self.config.environments:
            logger.error(f"Environment '{name}' already exists")
            return False

        # Create environment
        env = EnvironmentConfig(name=name, region=region, created_at=datetime.now())

        # Add to config
        self.config.environments[name] = env

        # Save config
        self.save_config()

        logger.info(f"Environment '{name}' added")
        return True
    except Exception as e:
        logger.error(f"Failed to add environment: {str(e)}")
        return False

add_iam_role(name, role_config)

Add an IAM role to global resources.

Parameters:

Name Type Description Default
name str

Role name.

required
role_config IAMRoleConfig

Role configuration.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def add_iam_role(self, name: str, role_config: IAMRoleConfig) -> bool:
    """Add an IAM role to global resources.

    Args:
        name: Role name.
        role_config: Role configuration.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Set or update the IAM role in global resources
        self.config.global_resources.iam_roles[name] = role_config

        # Save config
        self.save_config()

        logger.info(f"IAM role '{name}' added to global resources")
        return True
    except Exception as e:
        logger.error(f"Failed to add IAM role: {str(e)}")
        return False

delete_agent_runtime(name, environment=None)

Delete an agent runtime from the specified or current environment.

Parameters:

Name Type Description Default
name str

Agent runtime name.

required
environment str | None

Environment name. If None, uses current environment.

None

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def delete_agent_runtime(self, name: str, environment: str | None = None) -> bool:
    """Delete an agent runtime from the specified or current environment.

    Args:
        name: Agent runtime name.
        environment: Environment name. If None, uses current environment.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        env_name = environment or self.config.current_environment
        env = self.get_environment(env_name)

        # Check if agent runtime exists in this environment
        if name not in env.agent_runtimes:
            logger.error(f"Agent runtime '{name}' does not exist in environment '{env_name}'")
            return False

        # Delete agent runtime
        del env.agent_runtimes[name]

        # Update environment timestamp
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Agent runtime '{name}' deleted from environment '{env_name}'")
        return True
    except KeyError:
        logger.error(f"Environment '{env_name}' does not exist")
        return False
    except Exception as e:
        logger.error(f"Failed to delete agent runtime: {str(e)}")
        return False

delete_environment(name)

Delete an environment.

Parameters:

Name Type Description Default
name str

Environment name.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def delete_environment(self, name: str) -> bool:
    """Delete an environment.

    Args:
        name: Environment name.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Check if environment exists
        if name not in self.config.environments:
            logger.error(f"Environment '{name}' does not exist")
            return False

        # Check if it's the current environment
        if name == self.config.current_environment:
            logger.error(f"Cannot delete current environment '{name}'")
            return False

        # Delete environment
        del self.config.environments[name]

        # Save config
        self.save_config()

        logger.info(f"Environment '{name}' deleted")
        return True
    except Exception as e:
        logger.error(f"Failed to delete environment: {str(e)}")
        return False

enable_auto_sync(enable=True)

Enable or disable automatic configuration sync.

Parameters:

Name Type Description Default
enable bool

Whether to enable or disable auto-sync.

True

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def enable_auto_sync(self, enable: bool = True) -> bool:
    """Enable or disable automatic configuration sync.

    Args:
        enable: Whether to enable or disable auto-sync.

    Returns:
        bool: True if successful, False otherwise.
    """
    from agentcore_cli.services.config_sync import ConfigSyncService

    region = self.get_region()
    sync_service = ConfigSyncService(region=region, config=self.config)

    result = sync_service.enable_auto_sync(self.config, enable)

    if result:
        self.save_config()

    return result

enable_cloud_sync(enable=True)

Enable or disable cloud configuration sync.

Parameters:

Name Type Description Default
enable bool

Whether to enable or disable cloud sync.

True

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def enable_cloud_sync(self, enable: bool = True) -> bool:
    """Enable or disable cloud configuration sync.

    Args:
        enable: Whether to enable or disable cloud sync.

    Returns:
        bool: True if successful, False otherwise.
    """
    from agentcore_cli.services.config_sync import ConfigSyncService

    region = self.get_region()
    sync_service = ConfigSyncService(region=region, config=self.config)

    result = sync_service.enable_cloud_sync(self.config, enable)

    if result:
        self.save_config()

    return result

export_config(file_path)

Export configuration to a file.

Parameters:

Name Type Description Default
file_path str

Path to export the configuration to.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def export_config(self, file_path: str) -> bool:
    """Export configuration to a file.

    Args:
        file_path: Path to export the configuration to.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Convert to dict
        config_dict = self.config.model_dump(mode="json")

        # Remove config_path to avoid circular references
        if "config_path" in config_dict:
            del config_dict["config_path"]

        # Write to file
        with open(file_path, "w") as f:
            json.dump(config_dict, f, indent=2, default=str)

        logger.info(f"Configuration exported to {file_path}")
        return True
    except Exception as e:
        logger.error(f"Failed to export configuration: {str(e)}")
        return False

get_agent_runtime(name, environment=None)

Get an agent runtime by name from the specified or current environment.

Parameters:

Name Type Description Default
name str

Agent runtime name.

required
environment str | None

Environment name. If None, uses current environment.

None

Returns:

Type Description
AgentRuntime | None

Optional[AgentRuntime]: Agent runtime if found, None otherwise.

Source code in agentcore_cli/services/config.py
Python
def get_agent_runtime(self, name: str, environment: str | None = None) -> AgentRuntime | None:
    """Get an agent runtime by name from the specified or current environment.

    Args:
        name: Agent runtime name.
        environment: Environment name. If None, uses current environment.

    Returns:
        Optional[AgentRuntime]: Agent runtime if found, None otherwise.
    """
    env_name = environment or self.config.current_environment
    try:
        env = self.get_environment(env_name)
        return env.agent_runtimes.get(name)
    except KeyError:
        logger.error(f"Environment '{env_name}' does not exist")
        return None

get_cloud_sync_status(environment=None)

Get the sync status between local and cloud configuration.

Parameters:

Name Type Description Default
environment str | None

Optional environment to check. If None, checks current environment.

None

Returns:

Name Type Description
SyncStatus SyncStatus

Sync status between local and cloud.

Source code in agentcore_cli/services/config.py
Python
def get_cloud_sync_status(self, environment: str | None = None) -> SyncStatus:
    """Get the sync status between local and cloud configuration.

    Args:
        environment: Optional environment to check. If None, checks current environment.

    Returns:
        SyncStatus: Sync status between local and cloud.
    """
    from agentcore_cli.services.config_sync import ConfigSyncService

    env = environment or self.config.current_environment
    region = self.get_environment(env).region

    sync_service = ConfigSyncService(region=region, config=self.config)
    return sync_service.check_sync_status(self.config, env)

get_environment(name=None)

Get an environment configuration.

Parameters:

Name Type Description Default
name str | None

Optional environment name. If None, uses current environment.

None

Returns:

Name Type Description
EnvironmentConfig EnvironmentConfig

Environment configuration.

Raises:

Type Description
KeyError

If the environment does not exist.

Source code in agentcore_cli/services/config.py
Python
def get_environment(self, name: str | None = None) -> EnvironmentConfig:
    """Get an environment configuration.

    Args:
        name: Optional environment name. If None, uses current environment.

    Returns:
        EnvironmentConfig: Environment configuration.

    Raises:
        KeyError: If the environment does not exist.
    """
    env_name = name or self.config.current_environment

    if env_name not in self.config.environments:
        raise KeyError(f"Environment '{env_name}' does not exist")

    return self.config.environments[env_name]

get_region(environment=None)

Get the AWS region for an environment.

Parameters:

Name Type Description Default
environment str | None

Optional environment name. If None, uses current environment.

None

Returns:

Name Type Description
str str

AWS region.

Source code in agentcore_cli/services/config.py
Python
def get_region(self, environment: str | None = None) -> str:
    """Get the AWS region for an environment.

    Args:
        environment: Optional environment name. If None, uses current environment.

    Returns:
        str: AWS region.
    """
    env_name = environment or self.config.current_environment

    if env_name in self.config.environments:
        return self.config.environments[env_name].region
    else:
        # Default to us-west-2
        return "us-west-2"

import_config(file_path)

Import configuration from a file.

Parameters:

Name Type Description Default
file_path str

Path to import the configuration from.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def import_config(self, file_path: str) -> bool:
    """Import configuration from a file.

    Args:
        file_path: Path to import the configuration from.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Read file
        with open(file_path, "r") as f:
            config_data = json.load(f)

        # Parse config
        new_config = AgentCoreConfig.model_validate(config_data)

        # Keep the config path
        new_config.config_path = self.config_file

        # Update config
        self.config = new_config

        # Save config
        self.save_config()

        logger.info(f"Configuration imported from {file_path}")
        return True
    except Exception as e:
        logger.error(f"Failed to import configuration: {str(e)}")
        return False

pull_from_cloud()

Pull configuration from AWS Parameter Store.

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def pull_from_cloud(self) -> bool:
    """Pull configuration from AWS Parameter Store.

    Returns:
        bool: True if successful, False otherwise.
    """
    from agentcore_cli.services.config_sync import ConfigSyncService

    # Only sync if cloud sync is enabled
    if (
        not self.config.global_resources.sync_config
        or not self.config.global_resources.sync_config.cloud_config_enabled
    ):
        logger.error("Cloud sync is not enabled")
        return False

    # Get current environment region
    region = self.get_region()

    # Create sync service
    sync_service = ConfigSyncService(region=region, config=self.config)

    # Pull config from cloud
    success, config_dict, errors = sync_service.pull_config_from_cloud()

    if not success:
        logger.error(f"Failed to pull configuration from cloud: {errors}")
        return False

    try:
        # Update config with cloud data
        self.config = AgentCoreConfig.model_validate(config_dict)
        self.config.config_path = self.config_file

        # Save the updated config
        self.save_config()

        return True
    except Exception as e:
        logger.error(f"Failed to update configuration with cloud data: {str(e)}")
        return False

save_config()

Save current configuration to file.

Source code in agentcore_cli/services/config.py
Python
def save_config(self) -> bool:
    """Save current configuration to file."""
    try:
        # Ensure the config directory exists
        os.makedirs(self.config_dir, exist_ok=True)

        # Convert to dict and save
        config_dict = self.config.model_dump(mode="json")

        # Remove config_path to avoid circular references
        if "config_path" in config_dict:
            del config_dict["config_path"]

        with open(self.config_file, "w") as f:
            json.dump(config_dict, f, indent=2, default=str)

        logger.debug(f"Configuration saved to {self.config_file}")

        # Perform auto-sync if enabled
        if (
            self.config.global_resources.sync_config
            and self.config.global_resources.sync_config.cloud_config_enabled
        ):
            self.sync_with_cloud(auto=True)

        return True
    except Exception as e:
        logger.error(f"Failed to save configuration: {str(e)}")
        return False

set_current_environment(env_name)

Set the current environment.

Parameters:

Name Type Description Default
env_name str

Environment name.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def set_current_environment(self, env_name: str) -> bool:
    """Set the current environment.

    Args:
        env_name: Environment name.

    Returns:
        bool: True if successful, False otherwise.
    """
    # Check if environment exists
    if env_name not in self.config.environments:
        logger.error(f"Environment '{env_name}' does not exist")
        return False

    # Set current environment
    self.config.current_environment = env_name
    self.save_config()

    logger.debug(f"Current environment set to '{env_name}'")
    return True

set_default_agent_runtime(env_name, agent_name)

Set the default agent runtime for an environment.

Parameters:

Name Type Description Default
env_name str

Environment name.

required
agent_name str

Agent runtime name.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def set_default_agent_runtime(self, env_name: str, agent_name: str) -> bool:
    """Set the default agent runtime for an environment.

    Args:
        env_name: Environment name.
        agent_name: Agent runtime name.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Check if environment exists
        if env_name not in self.config.environments:
            logger.error(f"Environment '{env_name}' does not exist")
            return False

        # Get environment
        env = self.config.environments[env_name]

        # Check if agent runtime exists in this environment
        if agent_name not in env.agent_runtimes:
            logger.error(f"Agent runtime '{agent_name}' does not exist in environment '{env_name}'")
            return False

        # Set default agent runtime
        env.default_agent_runtime = agent_name

        # Update timestamp
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Default agent runtime set to '{agent_name}' in environment '{env_name}'")
        return True
    except Exception as e:
        logger.error(f"Failed to set default agent runtime: {str(e)}")
        return False

set_sync_interval(minutes)

Set the sync interval for auto-sync.

Parameters:

Name Type Description Default
minutes int

Sync interval in minutes.

required

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def set_sync_interval(self, minutes: int) -> bool:
    """Set the sync interval for auto-sync.

    Args:
        minutes: Sync interval in minutes.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        if not self.config.global_resources.sync_config:
            self.config.global_resources.sync_config = SyncConfig()

        self.config.global_resources.sync_config.sync_interval_minutes = minutes
        self.save_config()

        return True
    except Exception as e:
        logger.error(f"Failed to set sync interval: {str(e)}")
        return False

sync_with_cloud(auto=False)

Sync configuration with AWS Parameter Store.

Parameters:

Name Type Description Default
auto bool

Whether this is an automatic sync.

False

Returns:

Name Type Description
CloudSyncResult CloudSyncResult

Result of the sync operation.

Source code in agentcore_cli/services/config.py
Python
def sync_with_cloud(self, auto: bool = False) -> CloudSyncResult:
    """Sync configuration with AWS Parameter Store.

    Args:
        auto: Whether this is an automatic sync.

    Returns:
        CloudSyncResult: Result of the sync operation.
    """
    from agentcore_cli.services.config_sync import ConfigSyncService

    # Only sync if cloud sync is enabled
    if not self.config.global_resources.sync_config.cloud_config_enabled:
        return CloudSyncResult(
            success=False,
            message="Cloud sync is not enabled",
            environment=self.config.current_environment,
            synced_items={},
            errors=["Cloud sync is not enabled. Enable it first with 'config sync --enable'."],
        )

    # Only auto-sync if auto-sync is enabled
    if auto and (not self.config.global_resources.sync_config.auto_sync_enabled):
        return CloudSyncResult(
            success=True,
            message="Auto-sync is disabled",
            environment=self.config.current_environment,
            synced_items={},
            errors=[],
        )

    # Get current environment region
    region = self.get_region()

    # Create sync service
    sync_service = ConfigSyncService(region=region, config=self.config)

    # Check if we should auto-sync
    if auto and not sync_service.should_auto_sync:
        return CloudSyncResult(
            success=True,
            message="Automatic sync not needed",
            environment=self.config.current_environment,
            synced_items={},
            errors=[],
        )

    # Push config to cloud
    return sync_service.push_config_to_cloud(self.config)

update_agent_runtime(name, environment=None, **kwargs)

Update an agent runtime in the specified or current environment.

Parameters:

Name Type Description Default
name str

Agent runtime name.

required
environment str | None

Environment name. If None, uses current environment.

None
**kwargs Any

Runtime properties to update.

{}

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def update_agent_runtime(self, name: str, environment: str | None = None, **kwargs: Any) -> bool:
    """Update an agent runtime in the specified or current environment.

    Args:
        name: Agent runtime name.
        environment: Environment name. If None, uses current environment.
        **kwargs: Runtime properties to update.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        env_name = environment or self.config.current_environment
        env = self.get_environment(env_name)

        # Check if agent runtime exists in this environment
        if name not in env.agent_runtimes:
            logger.error(f"Agent runtime '{name}' does not exist in environment '{env_name}'")
            return False

        # Get agent runtime
        runtime = env.agent_runtimes[name]

        # Update properties
        for key, value in kwargs.items():
            if hasattr(runtime, key):
                setattr(runtime, key, value)

        # Update timestamp
        runtime.updated_at = datetime.now()
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Agent runtime '{name}' updated in environment '{env_name}'")
        return True
    except KeyError:
        logger.error(f"Environment '{env_name}' does not exist")
        return False
    except Exception as e:
        logger.error(f"Failed to update agent runtime: {str(e)}")
        return False

update_environment(name, **kwargs)

Update an environment.

Parameters:

Name Type Description Default
name str

Environment name.

required
**kwargs Any

Environment properties to update.

{}

Returns:

Name Type Description
bool bool

True if successful, False otherwise.

Source code in agentcore_cli/services/config.py
Python
def update_environment(self, name: str, **kwargs: Any) -> bool:
    """Update an environment.

    Args:
        name: Environment name.
        **kwargs: Environment properties to update.

    Returns:
        bool: True if successful, False otherwise.
    """
    try:
        # Check if environment exists
        if name not in self.config.environments:
            logger.error(f"Environment '{name}' does not exist")
            return False

        # Get environment
        env = self.config.environments[name]

        # Update properties
        for key, value in kwargs.items():
            if hasattr(env, key):
                setattr(env, key, value)

        # Update timestamp
        env.updated_at = datetime.now()

        # Save config
        self.save_config()

        logger.info(f"Environment '{name}' updated")
        return True
    except Exception as e:
        logger.error(f"Failed to update environment: {str(e)}")
        return False