agentcore_cli.models

agentcore_cli.models

Models package for AgentCore Platform CLI.

This package provides data models for the AgentCore Platform CLI.

ActionResult

Bases: BaseAgentCoreModel

Base model for CLI action results.

Source code in agentcore_cli/models/responses.py

class ActionResult(BaseAgentCoreModel):
    """Base model for CLI action results."""

    success: bool = Field(description="Whether the action was successful")
    message: str = Field(description="Status message")

AgentCoreConfig

Bases: BaseAgentCoreModel

Root configuration for AgentCore CLI.

This configuration uses an environment-first approach where each environment owns its agent runtimes. This design aligns with AWS AgentCore where: - Runtimes exist in specific regions - Environments represent deployment targets (dev/staging/prod) - Each environment can have different versions and endpoints

Source code in agentcore_cli/models/config.py

class AgentCoreConfig(BaseAgentCoreModel):
    """Root configuration for AgentCore CLI.

    This configuration uses an environment-first approach where each environment
    owns its agent runtimes. This design aligns with AWS AgentCore where:
    - Runtimes exist in specific regions
    - Environments represent deployment targets (dev/staging/prod)
    - Each environment can have different versions and endpoints
    """

    # Environment management
    current_environment: str = Field(default="dev", description="Currently active environment")
    environments: dict[str, EnvironmentConfig] = Field(
        default_factory=dict, description="Environment configurations by name"
    )

    # Global resources (shared across environments)
    global_resources: GlobalResourceConfig = Field(
        default_factory=GlobalResourceConfig, description="Resources shared across environments"
    )

    # File path for configuration
    config_path: str | None = Field(default=None, description="Path to the configuration file")

    model_config = ConfigDict(extra="allow", json_encoders={datetime: lambda v: v.isoformat() if v else None})

    @model_validator(mode="after")
    def ensure_current_environment_exists(cls, model):
        """Ensure the current environment exists."""
        if model.current_environment not in model.environments:
            # Create default environment
            model.environments[model.current_environment] = EnvironmentConfig(
                name=model.current_environment,
                region="us-east-1",  # Default region
                created_at=datetime.now(),
            )
        return model

    @model_validator(mode="after")
    def validate_ecr_repository_references(cls, model):
        """Validate that all ECR repository references exist in global_resources.ecr_repositories."""
        available_repos = set(model.global_resources.ecr_repositories.keys())

        for env_name, env_config in model.environments.items():
            for runtime_name, runtime in env_config.agent_runtimes.items():
                # Validate primary ECR repository
                if runtime.primary_ecr_repository not in available_repos:
                    raise ValueError(
                        f"Runtime '{runtime_name}' in environment '{env_name}' references "
                        f"ECR repository '{runtime.primary_ecr_repository}' which does not exist "
                        f"in global_resources.ecr_repositories. Available repositories: {sorted(available_repos)}"
                    )

                # Validate ECR repository references in all versions
                for version_id, version in runtime.versions.items():
                    if version.ecr_repository_name not in available_repos:
                        raise ValueError(
                            f"Version '{version_id}' of runtime '{runtime_name}' in environment '{env_name}' "
                            f"references ECR repository '{version.ecr_repository_name}' which does not exist "
                            f"in global_resources.ecr_repositories. Available repositories: {sorted(available_repos)}"
                        )

        return model

    def get_current_env(self) -> EnvironmentConfig:
        """Get the currently active environment configuration."""
        return self.environments[self.current_environment]

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

        Args:
            agent_name: Name of the agent runtime
            environment: Environment name, or None for current environment

        Returns:
            AgentRuntime instance or None if not found
        """
        env_name = environment or self.current_environment
        if env_name not in self.environments:
            return None

        env_config = self.environments[env_name]
        return env_config.agent_runtimes.get(agent_name)

    def list_all_agent_runtimes(self) -> dict[str, list[str]]:
        """List all agent runtimes across all environments.

        Returns:
            Dict mapping environment names to lists of agent runtime names
        """
        return {env_name: list(env_config.agent_runtimes.keys()) for env_name, env_config in self.environments.items()}

    def get_ecr_repository(self, repository_name: str) -> "ECRRepository | None":
        """Get an ECR repository by name.

        Args:
            repository_name: Name of the ECR repository

        Returns:
            ECRRepository instance or None if not found
        """
        # Validate repository name
        is_valid, error_msg = validate_repo_name(repository_name)
        if not is_valid:
            # Log warning but don't raise exception to maintain backward compatibility
            from loguru import logger

            logger.warning(f"Invalid repository name '{repository_name}': {error_msg}")
            return None

        return self.global_resources.ecr_repositories.get(repository_name)

    def get_runtime_version_container_uri(
        self, agent_name: str, version_id: str, environment: str | None = None
    ) -> str | None:
        """Get the full container URI for a specific runtime version.

        Args:
            agent_name: Name of the agent runtime
            version_id: Version identifier
            environment: Environment name, or None for current environment

        Returns:
            Full container URI or None if not found
        """
        runtime = self.get_agent_runtime(agent_name, environment)
        if not runtime or version_id not in runtime.versions:
            return None

        version = runtime.versions[version_id]
        ecr_repo = self.get_ecr_repository(version.ecr_repository_name)
        if not ecr_repo:
            return None

        return version.get_container_uri(ecr_repo)

ensure_current_environment_exists(model)

Ensure the current environment exists.

Source code in agentcore_cli/models/config.py

@model_validator(mode="after")
def ensure_current_environment_exists(cls, model):
    """Ensure the current environment exists."""
    if model.current_environment not in model.environments:
        # Create default environment
        model.environments[model.current_environment] = EnvironmentConfig(
            name=model.current_environment,
            region="us-east-1",  # Default region
            created_at=datetime.now(),
        )
    return model

get_agent_runtime(agent_name, environment=None)

Get an agent runtime from the specified or current environment.

Parameters:

Name	Type	Description	Default
agent_name	str	Name of the agent runtime	required
environment	str | None	Environment name, or None for current environment	None

Returns:

Type	Description
AgentRuntime | None	AgentRuntime instance or None if not found

Source code in agentcore_cli/models/config.py

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

    Args:
        agent_name: Name of the agent runtime
        environment: Environment name, or None for current environment

    Returns:
        AgentRuntime instance or None if not found
    """
    env_name = environment or self.current_environment
    if env_name not in self.environments:
        return None

    env_config = self.environments[env_name]
    return env_config.agent_runtimes.get(agent_name)

get_current_env()

Get the currently active environment configuration.

Source code in agentcore_cli/models/config.py

def get_current_env(self) -> EnvironmentConfig:
    """Get the currently active environment configuration."""
    return self.environments[self.current_environment]

get_ecr_repository(repository_name)

Get an ECR repository by name.

Parameters:

Name	Type	Description	Default
repository_name	str	Name of the ECR repository	required

Returns:

Type	Description
ECRRepository | None	ECRRepository instance or None if not found

Source code in agentcore_cli/models/config.py

def get_ecr_repository(self, repository_name: str) -> "ECRRepository | None":
    """Get an ECR repository by name.

    Args:
        repository_name: Name of the ECR repository

    Returns:
        ECRRepository instance or None if not found
    """
    # Validate repository name
    is_valid, error_msg = validate_repo_name(repository_name)
    if not is_valid:
        # Log warning but don't raise exception to maintain backward compatibility
        from loguru import logger

        logger.warning(f"Invalid repository name '{repository_name}': {error_msg}")
        return None

    return self.global_resources.ecr_repositories.get(repository_name)

get_runtime_version_container_uri(agent_name, version_id, environment=None)

Get the full container URI for a specific runtime version.

Parameters:

Name	Type	Description	Default
agent_name	str	Name of the agent runtime	required
version_id	str	Version identifier	required
environment	str | None	Environment name, or None for current environment	None

Returns:

Type	Description
str | None	Full container URI or None if not found

Source code in agentcore_cli/models/config.py

def get_runtime_version_container_uri(
    self, agent_name: str, version_id: str, environment: str | None = None
) -> str | None:
    """Get the full container URI for a specific runtime version.

    Args:
        agent_name: Name of the agent runtime
        version_id: Version identifier
        environment: Environment name, or None for current environment

    Returns:
        Full container URI or None if not found
    """
    runtime = self.get_agent_runtime(agent_name, environment)
    if not runtime or version_id not in runtime.versions:
        return None

    version = runtime.versions[version_id]
    ecr_repo = self.get_ecr_repository(version.ecr_repository_name)
    if not ecr_repo:
        return None

    return version.get_container_uri(ecr_repo)

list_all_agent_runtimes()

List all agent runtimes across all environments.

Returns:

Type	Description
dict[str, list[str]]	Dict mapping environment names to lists of agent runtime names

Source code in agentcore_cli/models/config.py

def list_all_agent_runtimes(self) -> dict[str, list[str]]:
    """List all agent runtimes across all environments.

    Returns:
        Dict mapping environment names to lists of agent runtime names
    """
    return {env_name: list(env_config.agent_runtimes.keys()) for env_name, env_config in self.environments.items()}

validate_ecr_repository_references(model)

Validate that all ECR repository references exist in global_resources.ecr_repositories.

Source code in agentcore_cli/models/config.py

@model_validator(mode="after")
def validate_ecr_repository_references(cls, model):
    """Validate that all ECR repository references exist in global_resources.ecr_repositories."""
    available_repos = set(model.global_resources.ecr_repositories.keys())

    for env_name, env_config in model.environments.items():
        for runtime_name, runtime in env_config.agent_runtimes.items():
            # Validate primary ECR repository
            if runtime.primary_ecr_repository not in available_repos:
                raise ValueError(
                    f"Runtime '{runtime_name}' in environment '{env_name}' references "
                    f"ECR repository '{runtime.primary_ecr_repository}' which does not exist "
                    f"in global_resources.ecr_repositories. Available repositories: {sorted(available_repos)}"
                )

            # Validate ECR repository references in all versions
            for version_id, version in runtime.versions.items():
                if version.ecr_repository_name not in available_repos:
                    raise ValueError(
                        f"Version '{version_id}' of runtime '{runtime_name}' in environment '{env_name}' "
                        f"references ECR repository '{version.ecr_repository_name}' which does not exist "
                        f"in global_resources.ecr_repositories. Available repositories: {sorted(available_repos)}"
                    )

    return model

AgentCreationResult

Bases: ActionResult

Result of agent creation.

Source code in agentcore_cli/models/responses.py

class AgentCreationResult(ActionResult):
    """Result of agent creation."""

    agent_name: str = Field(description="Agent name")
    runtime_id: str | None = Field(default=None, description="Runtime ID")
    runtime_arn: str | None = Field(default=None, description="Runtime ARN")
    container_uri: str | None = Field(default=None, description="Container URI")
    role_arn: str | None = Field(default=None, description="IAM role ARN")
    environment: str = Field(description="Environment name")
    default_endpoint: str = Field(default="DEFAULT", description="Default endpoint name")

AgentDeletionResult

Bases: ActionResult

Result of agent deletion.

Source code in agentcore_cli/models/responses.py

class AgentDeletionResult(ActionResult):
    """Result of agent deletion."""

    agent_name: str = Field(description="Agent name")
    deleted_resources: list[str] = Field(default_factory=list, description="Deleted resources")
    environment: str = Field(description="Environment name")

AgentEndpointStatusType

Bases: StrEnum

Status of an agent runtime endpoint.

Endpoints go through these states during their lifecycle. Unlike runtimes, endpoints can be updated to point to different versions.

Source code in agentcore_cli/models/base.py

class AgentEndpointStatusType(StrEnum):
    """Status of an agent runtime endpoint.

    Endpoints go through these states during their lifecycle.
    Unlike runtimes, endpoints can be updated to point to different versions.
    """

    CREATING = "CREATING"
    CREATE_FAILED = "CREATE_FAILED"
    UPDATING = "UPDATING"
    UPDATE_FAILED = "UPDATE_FAILED"
    READY = "READY"
    DELETING = "DELETING"

AgentInvocationResponse

Bases: BaseAgentCoreModel

Response model for agent invocation.

This model handles both streaming and non-streaming responses from the boto3 invoke_agent_runtime API.

Source code in agentcore_cli/models/responses.py

class AgentInvocationResponse(BaseAgentCoreModel):
    """Response model for agent invocation.

    This model handles both streaming and non-streaming responses from
    the boto3 invoke_agent_runtime API.
    """

    content_type: str | None = Field(default=None, description="Content type of the response")
    streaming: bool = Field(default=False, description="Whether response was streaming")
    session_id: str = Field(description="Session ID used for the invocation")
    agent_name: str = Field(description="Agent name that was invoked")
    endpoint_name: str = Field(description="Endpoint name that was used")
    timestamp: datetime = Field(default_factory=datetime.now, description="Timestamp of the invocation")

    # For streaming responses
    stream_content: list[str] | None = Field(default=None, description="Collected content from streaming response")

    # For non-streaming responses
    output: Any | None = Field(default=None, description="Response output object for non-streaming response")

    @classmethod
    def from_streaming_response(
        cls, response: Any, agent_name: str, endpoint_name: str, session_id: str
    ) -> "AgentInvocationResponse":
        """Create a response object from a streaming API response."""
        content = []

        # Handle streaming response
        if "response" in response and hasattr(response["response"], "iter_lines"):
            for line in response["response"].iter_lines(chunk_size=1):
                if line:
                    line_text = line.decode("utf-8")
                    # Check for "data: " prefix
                    if line_text.startswith("data: "):
                        line_text = line_text[6:]  # Remove the "data: " prefix
                        content.append(line_text)

        return cls(
            content_type=response.get("contentType"),
            streaming=True,
            session_id=session_id,
            agent_name=agent_name,
            endpoint_name=endpoint_name,
            stream_content=content,
        )

    @classmethod
    def from_nonstreaming_response(
        cls, response: Any, agent_name: str, endpoint_name: str, session_id: str
    ) -> "AgentInvocationResponse":
        """Create a response object from a non-streaming API response."""
        try:
            # Extract response body
            response_body = None
            if "response" in response:
                if hasattr(response["response"], "read"):
                    response_body = response["response"].read()
                    if response_body:
                        response_data = json.loads(response_body)
                        output = response_data.get("output", {})
                    else:
                        output = {}
                else:
                    output = response.get("response", {})
            else:
                output = {}

            return cls(
                content_type=response.get("contentType"),
                streaming=False,
                session_id=session_id,
                agent_name=agent_name,
                endpoint_name=endpoint_name,
                output=output,
            )

        except Exception as e:
            # Handle any parsing errors
            return cls(
                content_type=response.get("contentType"),
                streaming=False,
                session_id=session_id,
                agent_name=agent_name,
                endpoint_name=endpoint_name,
                output={"error": f"Failed to parse response: {str(e)}"},
            )

from_nonstreaming_response(response, agent_name, endpoint_name, session_id) classmethod

Create a response object from a non-streaming API response.

Source code in agentcore_cli/models/responses.py

@classmethod
def from_nonstreaming_response(
    cls, response: Any, agent_name: str, endpoint_name: str, session_id: str
) -> "AgentInvocationResponse":
    """Create a response object from a non-streaming API response."""
    try:
        # Extract response body
        response_body = None
        if "response" in response:
            if hasattr(response["response"], "read"):
                response_body = response["response"].read()
                if response_body:
                    response_data = json.loads(response_body)
                    output = response_data.get("output", {})
                else:
                    output = {}
            else:
                output = response.get("response", {})
        else:
            output = {}

        return cls(
            content_type=response.get("contentType"),
            streaming=False,
            session_id=session_id,
            agent_name=agent_name,
            endpoint_name=endpoint_name,
            output=output,
        )

    except Exception as e:
        # Handle any parsing errors
        return cls(
            content_type=response.get("contentType"),
            streaming=False,
            session_id=session_id,
            agent_name=agent_name,
            endpoint_name=endpoint_name,
            output={"error": f"Failed to parse response: {str(e)}"},
        )

from_streaming_response(response, agent_name, endpoint_name, session_id) classmethod

Create a response object from a streaming API response.

Source code in agentcore_cli/models/responses.py

@classmethod
def from_streaming_response(
    cls, response: Any, agent_name: str, endpoint_name: str, session_id: str
) -> "AgentInvocationResponse":
    """Create a response object from a streaming API response."""
    content = []

    # Handle streaming response
    if "response" in response and hasattr(response["response"], "iter_lines"):
        for line in response["response"].iter_lines(chunk_size=1):
            if line:
                line_text = line.decode("utf-8")
                # Check for "data: " prefix
                if line_text.startswith("data: "):
                    line_text = line_text[6:]  # Remove the "data: " prefix
                    content.append(line_text)

    return cls(
        content_type=response.get("contentType"),
        streaming=True,
        session_id=session_id,
        agent_name=agent_name,
        endpoint_name=endpoint_name,
        stream_content=content,
    )

AgentRuntime

Bases: ResourceBase

Configuration for an agent runtime.

Agent runtimes are associated with ECR repositories where their container images are stored. Each runtime can have multiple versions, each pointing to different image tags.

Source code in agentcore_cli/models/runtime.py

class AgentRuntime(ResourceBase):
    """Configuration for an agent runtime.

    Agent runtimes are associated with ECR repositories where their container images are stored.
    Each runtime can have multiple versions, each pointing to different image tags.
    """

    name: str = Field(description="Agent runtime name")
    agent_runtime_id: str = Field(description="Runtime ID")
    agent_runtime_arn: str | None = Field(default=None, description="Runtime ARN")
    description: str | None = Field(default=None, description="Runtime description")
    latest_version: str = Field(description="Latest version identifier")

    # ECR Repository association
    primary_ecr_repository: str = Field(
        description="Primary ECR repository name for this runtime (must exist in global_resources.ecr_repositories)"
    )

    # Runtime components
    versions: dict[str, AgentRuntimeVersion] = Field(
        default_factory=dict, description="Available versions by version identifier"
    )
    endpoints: dict[str, AgentRuntimeEndpoint] = Field(default_factory=dict, description="Available endpoints by name")
    workload_identity: WorkloadIdentity | None = Field(default=None, description="Workload identity details")
    authorizer_config: AuthorizerConfig | None = Field(default=None, description="Authorizer configuration")

    @field_validator("name")
    @classmethod
    def validate_runtime_name(cls, v: str) -> str:
        """Validate agent runtime name format."""
        if not v:
            raise ValueError("Agent runtime name cannot be empty")

        if not v[0].isalpha():
            raise ValueError("Agent runtime name must start with a letter")

        if not all(c.isalnum() or c == "-" for c in v):
            raise ValueError("Agent runtime name can only contain alphanumeric characters and hyphens")

        if len(v) < 3 or len(v) > 63:
            raise ValueError("Agent runtime name must be between 3 and 63 characters")

        return v

    @model_validator(mode="after")
    def validate_version_repository_consistency(cls, model):
        """Validate that all versions reference valid ECR repositories."""
        for version_id, version in model.versions.items():
            # All versions should typically use the same repository as the runtime's primary repository
            # This is a soft validation - you could have versions in different repositories
            if version.ecr_repository_name != model.primary_ecr_repository:
                import warnings

                warnings.warn(
                    f"Version '{version_id}' uses repository '{version.ecr_repository_name}' "
                    f"but runtime '{model.name}' primary repository is '{model.primary_ecr_repository}'"
                )
        return model

    @model_validator(mode="after")
    def ensure_default_endpoint(cls, model):
        """Ensure the DEFAULT endpoint exists and points to latest version.

        Note: AWS AgentCore automatically creates and manages the DEFAULT endpoint.
        This validator ensures our model reflects this AWS behavior.
        """
        if model.versions and model.latest_version in model.versions:
            # If we have versions but no DEFAULT endpoint info, create a placeholder
            # representing what AWS would automatically create
            if "DEFAULT" not in model.endpoints:
                model.endpoints["DEFAULT"] = AgentRuntimeEndpoint(
                    name="DEFAULT",
                    agent_runtime_id=model.agent_runtime_id,
                    target_version=model.latest_version,
                    status=AgentEndpointStatusType.READY,
                    created_at=model.created_at,
                    description="Default endpoint (automatically managed by AWS)",
                )
            else:
                # Ensure DEFAULT endpoint points to latest version (AWS behavior)
                default_endpoint = model.endpoints["DEFAULT"]
                if default_endpoint.target_version != model.latest_version:
                    default_endpoint.target_version = model.latest_version
        return model

    def get_version_container_uri(self, version_id: str, ecr_repository: "ECRRepository") -> str | None:
        """Get the full container URI for a specific version.

        Args:
            version_id: Version identifier
            ecr_repository: ECR repository configuration

        Returns:
            Full container URI or None if version not found
        """
        if version_id not in self.versions:
            return None

        version = self.versions[version_id]
        return version.get_container_uri(ecr_repository)

ensure_default_endpoint(model)

Ensure the DEFAULT endpoint exists and points to latest version.

Note: AWS AgentCore automatically creates and manages the DEFAULT endpoint. This validator ensures our model reflects this AWS behavior.

Source code in agentcore_cli/models/runtime.py

@model_validator(mode="after")
def ensure_default_endpoint(cls, model):
    """Ensure the DEFAULT endpoint exists and points to latest version.

    Note: AWS AgentCore automatically creates and manages the DEFAULT endpoint.
    This validator ensures our model reflects this AWS behavior.
    """
    if model.versions and model.latest_version in model.versions:
        # If we have versions but no DEFAULT endpoint info, create a placeholder
        # representing what AWS would automatically create
        if "DEFAULT" not in model.endpoints:
            model.endpoints["DEFAULT"] = AgentRuntimeEndpoint(
                name="DEFAULT",
                agent_runtime_id=model.agent_runtime_id,
                target_version=model.latest_version,
                status=AgentEndpointStatusType.READY,
                created_at=model.created_at,
                description="Default endpoint (automatically managed by AWS)",
            )
        else:
            # Ensure DEFAULT endpoint points to latest version (AWS behavior)
            default_endpoint = model.endpoints["DEFAULT"]
            if default_endpoint.target_version != model.latest_version:
                default_endpoint.target_version = model.latest_version
    return model

get_version_container_uri(version_id, ecr_repository)

Get the full container URI for a specific version.

Parameters:

Name	Type	Description	Default
version_id	str	Version identifier	required
ecr_repository	ECRRepository	ECR repository configuration	required

Returns:

Type	Description
str | None	Full container URI or None if version not found

Source code in agentcore_cli/models/runtime.py

def get_version_container_uri(self, version_id: str, ecr_repository: "ECRRepository") -> str | None:
    """Get the full container URI for a specific version.

    Args:
        version_id: Version identifier
        ecr_repository: ECR repository configuration

    Returns:
        Full container URI or None if version not found
    """
    if version_id not in self.versions:
        return None

    version = self.versions[version_id]
    return version.get_container_uri(ecr_repository)

validate_runtime_name(v) classmethod

Validate agent runtime name format.

Source code in agentcore_cli/models/runtime.py

@field_validator("name")
@classmethod
def validate_runtime_name(cls, v: str) -> str:
    """Validate agent runtime name format."""
    if not v:
        raise ValueError("Agent runtime name cannot be empty")

    if not v[0].isalpha():
        raise ValueError("Agent runtime name must start with a letter")

    if not all(c.isalnum() or c == "-" for c in v):
        raise ValueError("Agent runtime name can only contain alphanumeric characters and hyphens")

    if len(v) < 3 or len(v) > 63:
        raise ValueError("Agent runtime name must be between 3 and 63 characters")

    return v

validate_version_repository_consistency(model)

Validate that all versions reference valid ECR repositories.

Source code in agentcore_cli/models/runtime.py

@model_validator(mode="after")
def validate_version_repository_consistency(cls, model):
    """Validate that all versions reference valid ECR repositories."""
    for version_id, version in model.versions.items():
        # All versions should typically use the same repository as the runtime's primary repository
        # This is a soft validation - you could have versions in different repositories
        if version.ecr_repository_name != model.primary_ecr_repository:
            import warnings

            warnings.warn(
                f"Version '{version_id}' uses repository '{version.ecr_repository_name}' "
                f"but runtime '{model.name}' primary repository is '{model.primary_ecr_repository}'"
            )
    return model

AgentRuntimeEndpoint

Bases: BaseAgentCoreModel

An endpoint for accessing a specific version of an agent runtime.

Source code in agentcore_cli/models/runtime.py

class AgentRuntimeEndpoint(BaseAgentCoreModel):
    """An endpoint for accessing a specific version of an agent runtime."""

    name: str = Field(description="Endpoint name (e.g., 'DEFAULT', 'prod', 'dev')")
    agent_runtime_id: str = Field(description="Runtime ID this endpoint belongs to")
    target_version: str = Field(description="Version identifier this endpoint points to")
    status: AgentEndpointStatusType = Field(description="Current endpoint status")
    description: str | None = Field(default=None, description="Endpoint description")
    created_at: datetime | None = Field(default=None, description="When this endpoint was created")
    updated_at: datetime | None = Field(default=None, description="When this endpoint was last updated")
    endpoint_arn: str | None = Field(default=None, description="ARN of this endpoint")
    failure_reason: str | None = Field(
        default=None, description="Reason for failure if status is CREATE_FAILED or UPDATE_FAILED"
    )
    live_version: str | None = Field(default=None, description="Currently active version for this endpoint")

    @field_validator("name")
    @classmethod
    def validate_endpoint_name(cls, v: str) -> str:
        """Validate endpoint name format."""
        if not v:
            raise ValueError("Endpoint name cannot be empty")

        if not v.isalnum() and not all(c.isalnum() or c == "-" for c in v):
            raise ValueError("Endpoint name can only contain alphanumeric characters and hyphens")

        return v

validate_endpoint_name(v) classmethod

Validate endpoint name format.

Source code in agentcore_cli/models/runtime.py

@field_validator("name")
@classmethod
def validate_endpoint_name(cls, v: str) -> str:
    """Validate endpoint name format."""
    if not v:
        raise ValueError("Endpoint name cannot be empty")

    if not v.isalnum() and not all(c.isalnum() or c == "-" for c in v):
        raise ValueError("Endpoint name can only contain alphanumeric characters and hyphens")

    return v

AgentRuntimeResponseAdapter

Helper class to convert AWS API responses to our model structures.

Source code in agentcore_cli/models/adapters.py

class AgentRuntimeResponseAdapter:
    """Helper class to convert AWS API responses to our model structures."""

    @staticmethod
    def from_aws_response(response: Any) -> AgentRuntime:
        """Convert AWS API response to AgentRuntime model."""
        # Extract container URI
        container_uri = response.get("agentRuntimeArtifact", {}).get("containerConfiguration", {}).get("containerUri")

        # Determine network mode
        network_mode = NetworkModeType(
            response.get("networkConfiguration", {}).get("networkMode", NetworkModeType.PUBLIC)
        )

        # Determine protocol
        protocol = ServerProtocolType(
            response.get("protocolConfiguration", {}).get("serverProtocol", ServerProtocolType.HTTP)
        )

        # Extract region from ARN if available
        region = "us-east-1"  # Default region
        if "agentRuntimeArn" in response and response["agentRuntimeArn"]:
            arn_parts = response["agentRuntimeArn"].split(":")
            if len(arn_parts) >= 4:
                region = arn_parts[3]

        # Extract workload identity
        workload_identity = None
        if "workloadIdentityDetails" in response and "workloadIdentityArn" in response["workloadIdentityDetails"]:
            workload_identity = WorkloadIdentity(
                workload_identity_arn=response["workloadIdentityDetails"]["workloadIdentityArn"]
            )

        # Determine status
        status = AgentStatusType(response.get("status", AgentStatusType.CREATING))

        # Parse timestamps
        created_at = response.get("createdAt")
        updated_at = response.get("updatedAt")

        # Extract version info
        version_id = "V1"  # Default to V1 (AWS format)
        if "agentRuntimeVersion" in response:
            raw_version = response["agentRuntimeVersion"]
            # Ensure consistent format - AWS uses V1, V2, etc.
            if raw_version and not raw_version.upper().startswith("V"):
                version_id = f"V{raw_version}"
            else:
                version_id = raw_version.upper() if raw_version else "V1"

        # Parse ECR information from container URI
        ecr_repository_name = ""
        image_tag = "latest"

        if container_uri:
            # Parse container URI: registry/repository:tag
            if ":" in container_uri and "/" in container_uri:
                repo_with_tag = container_uri.split("/")[-1]  # Get last part after /
                if ":" in repo_with_tag:
                    repo_name, tag = repo_with_tag.split(":", 1)
                    ecr_repository_name = repo_name
                    image_tag = tag
                else:
                    ecr_repository_name = repo_with_tag
            elif "/" in container_uri:
                # No tag specified, extract repository name
                ecr_repository_name = container_uri.split("/")[-1]

        # Create runtime version
        runtime_version = AgentRuntimeVersion(
            version_id=version_id,
            agent_runtime_id=response.get("agentRuntimeId", ""),
            ecr_repository_name=ecr_repository_name,
            image_tag=image_tag,
            status=status,
            created_at=created_at,
            network_mode=network_mode,
            protocol=protocol,
            environment_variables=response.get("environmentVariables", {}),
            execution_role_arn=response.get("roleArn", ""),
            failure_reason=response.get("failureReason"),
            updated_at=updated_at,
        )

        # Create default endpoint
        default_endpoint = AgentRuntimeEndpoint(
            name="DEFAULT",
            agent_runtime_id=response.get("agentRuntimeId", ""),
            target_version=version_id,
            status=AgentEndpointStatusType.READY,
            created_at=created_at,
            endpoint_arn=f"{response.get('agentRuntimeArn', '')}/endpoints/DEFAULT"
            if response.get("agentRuntimeArn")
            else None,
        )

        # Create and return AgentRuntime
        return AgentRuntime(
            name=response.get("agentRuntimeName", ""),
            agent_runtime_id=response.get("agentRuntimeId", ""),
            agent_runtime_arn=response.get("agentRuntimeArn"),
            description=response.get("description"),
            latest_version=version_id,
            primary_ecr_repository=ecr_repository_name,
            versions={version_id: runtime_version},
            endpoints={"DEFAULT": default_endpoint},
            created_at=created_at,
            updated_at=updated_at,
            region=region,
            tags=response.get("tags", {}),
            workload_identity=workload_identity,
        )

    @staticmethod
    def from_endpoint_response(
        response: AgentEndpointTypeDef | GetAgentRuntimeEndpointResponseTypeDef, agent_runtime_id: str
    ) -> AgentRuntimeEndpoint:
        """Convert AWS API endpoint response to AgentRuntimeEndpoint model."""
        status = AgentEndpointStatusType(response.get("status", AgentEndpointStatusType.READY))

        return AgentRuntimeEndpoint(
            name=response["name"],
            agent_runtime_id=agent_runtime_id,
            target_version=response.get("targetVersion", ""),
            status=status,
            description=response.get("description", ""),
            created_at=response["createdAt"],
            # Handle potential datetime conversion
            updated_at=response["lastUpdatedAt"],
            endpoint_arn=response.get("agentRuntimeEndpointArn"),
            failure_reason=str(response.get("failureReason")) if response.get("failureReason") else None,
            live_version=str(response.get("liveVersion")) if response.get("liveVersion") else None,
        )

from_aws_response(response) staticmethod

Convert AWS API response to AgentRuntime model.

Source code in agentcore_cli/models/adapters.py

@staticmethod
def from_aws_response(response: Any) -> AgentRuntime:
    """Convert AWS API response to AgentRuntime model."""
    # Extract container URI
    container_uri = response.get("agentRuntimeArtifact", {}).get("containerConfiguration", {}).get("containerUri")

    # Determine network mode
    network_mode = NetworkModeType(
        response.get("networkConfiguration", {}).get("networkMode", NetworkModeType.PUBLIC)
    )

    # Determine protocol
    protocol = ServerProtocolType(
        response.get("protocolConfiguration", {}).get("serverProtocol", ServerProtocolType.HTTP)
    )

    # Extract region from ARN if available
    region = "us-east-1"  # Default region
    if "agentRuntimeArn" in response and response["agentRuntimeArn"]:
        arn_parts = response["agentRuntimeArn"].split(":")
        if len(arn_parts) >= 4:
            region = arn_parts[3]

    # Extract workload identity
    workload_identity = None
    if "workloadIdentityDetails" in response and "workloadIdentityArn" in response["workloadIdentityDetails"]:
        workload_identity = WorkloadIdentity(
            workload_identity_arn=response["workloadIdentityDetails"]["workloadIdentityArn"]
        )

    # Determine status
    status = AgentStatusType(response.get("status", AgentStatusType.CREATING))

    # Parse timestamps
    created_at = response.get("createdAt")
    updated_at = response.get("updatedAt")

    # Extract version info
    version_id = "V1"  # Default to V1 (AWS format)
    if "agentRuntimeVersion" in response:
        raw_version = response["agentRuntimeVersion"]
        # Ensure consistent format - AWS uses V1, V2, etc.
        if raw_version and not raw_version.upper().startswith("V"):
            version_id = f"V{raw_version}"
        else:
            version_id = raw_version.upper() if raw_version else "V1"

    # Parse ECR information from container URI
    ecr_repository_name = ""
    image_tag = "latest"

    if container_uri:
        # Parse container URI: registry/repository:tag
        if ":" in container_uri and "/" in container_uri:
            repo_with_tag = container_uri.split("/")[-1]  # Get last part after /
            if ":" in repo_with_tag:
                repo_name, tag = repo_with_tag.split(":", 1)
                ecr_repository_name = repo_name
                image_tag = tag
            else:
                ecr_repository_name = repo_with_tag
        elif "/" in container_uri:
            # No tag specified, extract repository name
            ecr_repository_name = container_uri.split("/")[-1]

    # Create runtime version
    runtime_version = AgentRuntimeVersion(
        version_id=version_id,
        agent_runtime_id=response.get("agentRuntimeId", ""),
        ecr_repository_name=ecr_repository_name,
        image_tag=image_tag,
        status=status,
        created_at=created_at,
        network_mode=network_mode,
        protocol=protocol,
        environment_variables=response.get("environmentVariables", {}),
        execution_role_arn=response.get("roleArn", ""),
        failure_reason=response.get("failureReason"),
        updated_at=updated_at,
    )

    # Create default endpoint
    default_endpoint = AgentRuntimeEndpoint(
        name="DEFAULT",
        agent_runtime_id=response.get("agentRuntimeId", ""),
        target_version=version_id,
        status=AgentEndpointStatusType.READY,
        created_at=created_at,
        endpoint_arn=f"{response.get('agentRuntimeArn', '')}/endpoints/DEFAULT"
        if response.get("agentRuntimeArn")
        else None,
    )

    # Create and return AgentRuntime
    return AgentRuntime(
        name=response.get("agentRuntimeName", ""),
        agent_runtime_id=response.get("agentRuntimeId", ""),
        agent_runtime_arn=response.get("agentRuntimeArn"),
        description=response.get("description"),
        latest_version=version_id,
        primary_ecr_repository=ecr_repository_name,
        versions={version_id: runtime_version},
        endpoints={"DEFAULT": default_endpoint},
        created_at=created_at,
        updated_at=updated_at,
        region=region,
        tags=response.get("tags", {}),
        workload_identity=workload_identity,
    )

from_endpoint_response(response, agent_runtime_id) staticmethod

Convert AWS API endpoint response to AgentRuntimeEndpoint model.

Source code in agentcore_cli/models/adapters.py

@staticmethod
def from_endpoint_response(
    response: AgentEndpointTypeDef | GetAgentRuntimeEndpointResponseTypeDef, agent_runtime_id: str
) -> AgentRuntimeEndpoint:
    """Convert AWS API endpoint response to AgentRuntimeEndpoint model."""
    status = AgentEndpointStatusType(response.get("status", AgentEndpointStatusType.READY))

    return AgentRuntimeEndpoint(
        name=response["name"],
        agent_runtime_id=agent_runtime_id,
        target_version=response.get("targetVersion", ""),
        status=status,
        description=response.get("description", ""),
        created_at=response["createdAt"],
        # Handle potential datetime conversion
        updated_at=response["lastUpdatedAt"],
        endpoint_arn=response.get("agentRuntimeEndpointArn"),
        failure_reason=str(response.get("failureReason")) if response.get("failureReason") else None,
        live_version=str(response.get("liveVersion")) if response.get("liveVersion") else None,
    )

AgentRuntimeVersion

Bases: BaseAgentCoreModel

A specific version of an agent runtime.

Each version references a specific ECR repository and image tag. The container URI is constructed from the repository + tag combination.

Source code in agentcore_cli/models/runtime.py

class AgentRuntimeVersion(BaseAgentCoreModel):
    """A specific version of an agent runtime.

    Each version references a specific ECR repository and image tag.
    The container URI is constructed from the repository + tag combination.
    """

    version_id: str = Field(description="Version identifier (e.g., 'V1', 'V2')")
    agent_runtime_id: str = Field(description="Runtime ID this version belongs to")

    # ECR Image Reference
    ecr_repository_name: str = Field(
        description="Name of the ECR repository (must exist in global_resources.ecr_repositories)"
    )
    image_tag: str = Field(
        description="Docker image tag in the ECR repository (e.g., 'v1', 'latest', 'prod-2024-01-15')"
    )

    # Runtime Configuration
    status: AgentStatusType = Field(description="Status of this runtime version")
    created_at: datetime | None = Field(default=None, description="When this version was created")
    network_mode: NetworkModeType = Field(default=NetworkModeType.PUBLIC, description="Network mode for this version")
    protocol: ServerProtocolType = Field(default=ServerProtocolType.HTTP, description="Protocol type for this version")
    environment_variables: dict[str, str] = Field(
        default_factory=dict, description="Environment variables for this version"
    )
    execution_role_arn: str = Field(description="IAM role ARN used for this version")
    description: str | None = Field(default=None, description="Description of this version")
    failure_reason: str | None = Field(
        default=None, description="Reason for failure if status is CREATE_FAILED or UPDATE_FAILED"
    )
    updated_at: datetime | None = Field(default=None, description="When this version was last updated")

    model_config = ConfigDict(frozen=False)  # Allow updates for status changes, but versions are conceptually immutable

    @property
    def container_uri(self) -> str:
        """Get the full container URI by combining repository and tag.

        Note: This requires access to the ECR repository configuration to build the full URI.
        Use get_container_uri() method with repository config for the complete URI.
        """
        return f"<repository_uri>:{self.image_tag}"

    def get_container_uri(self, ecr_repository: "ECRRepository") -> str:
        """Get the full container URI using the ECR repository configuration.

        Args:
            ecr_repository: ECR repository configuration

        Returns:
            Full container URI: registry/repository:tag
        """
        return ecr_repository.get_image_uri(self.image_tag)

    @property
    def short_version(self) -> str:
        """Get a shortened version of the version ID for display."""
        # AWS AgentCore uses format like 'V1', 'V2', etc.
        # Handle various possible formats for backward compatibility
        if self.version_id.startswith("version-"):
            return "V" + self.version_id.replace("version-", "")
        elif self.version_id.lower().startswith("v"):
            return self.version_id.upper()
        else:
            # If it's just a number or other format, prefix with V
            return f"V{self.version_id}"

    @property
    def is_immutable(self) -> bool:
        """Check if this version is in an immutable state.

        According to AWS documentation, versions are immutable once created,
        but status can change during creation/update process.
        """
        return self.status in {AgentStatusType.READY, AgentStatusType.CREATE_FAILED, AgentStatusType.UPDATE_FAILED}

container_uri property

Get the full container URI by combining repository and tag.

Note: This requires access to the ECR repository configuration to build the full URI. Use get_container_uri() method with repository config for the complete URI.

is_immutable property

Check if this version is in an immutable state.

According to AWS documentation, versions are immutable once created, but status can change during creation/update process.

short_version property

Get a shortened version of the version ID for display.

get_container_uri(ecr_repository)

Get the full container URI using the ECR repository configuration.

Parameters:

Name	Type	Description	Default
ecr_repository	ECRRepository	ECR repository configuration	required

Returns:

Type	Description
str	Full container URI: registry/repository:tag

Source code in agentcore_cli/models/runtime.py

def get_container_uri(self, ecr_repository: "ECRRepository") -> str:
    """Get the full container URI using the ECR repository configuration.

    Args:
        ecr_repository: ECR repository configuration

    Returns:
        Full container URI: registry/repository:tag
    """
    return ecr_repository.get_image_uri(self.image_tag)

AgentStatusType

Bases: StrEnum

Status of an agent runtime.

These statuses reflect the lifecycle of an AWS AgentCore runtime. Runtimes transition through these states during creation, updates, and deletion.

Source code in agentcore_cli/models/base.py

class AgentStatusType(StrEnum):
    """Status of an agent runtime.

    These statuses reflect the lifecycle of an AWS AgentCore runtime.
    Runtimes transition through these states during creation, updates, and deletion.
    """

    CREATING = "CREATING"
    CREATE_FAILED = "CREATE_FAILED"
    UPDATING = "UPDATING"
    UPDATE_FAILED = "UPDATE_FAILED"
    READY = "READY"
    DELETING = "DELETING"

AgentUpdateResult

Bases: ActionResult

Result of agent update.

Source code in agentcore_cli/models/responses.py

class AgentUpdateResult(ActionResult):
    """Result of agent update."""

    agent_name: str = Field(description="Agent name")
    runtime_id: str = Field(description="Runtime ID")
    version: str = Field(description="New runtime version created")
    container_uri: str = Field(description="Container URI used")
    environment: str = Field(description="Environment name")

AuthorizerConfig

Bases: BaseAgentCoreModel

Authorizer configuration for AgentCore runtime.

Source code in agentcore_cli/models/runtime.py

class AuthorizerConfig(BaseAgentCoreModel):
    """Authorizer configuration for AgentCore runtime."""

    custom_jwt_authorizer: CustomJWTAuthorizer | None = Field(default=None, description="JWT authorizer configuration")

BaseAgentCoreModel

Bases: BaseModel

Base model with common configuration for all AgentCore models.

Provides strict validation and consistent behavior across all models. All AgentCore CLI models inherit from this base class.

Source code in agentcore_cli/models/base.py

class BaseAgentCoreModel(BaseModel):
    """Base model with common configuration for all AgentCore models.

    Provides strict validation and consistent behavior across all models.
    All AgentCore CLI models inherit from this base class.
    """

    model_config = ConfigDict(
        extra="forbid",  # Forbid extra attributes for strict AWS API alignment
        validate_default=True,  # Validate default values
        validate_assignment=True,  # Validate attribute assignments
        str_strip_whitespace=True,  # Strip whitespace from string values
        use_enum_values=True,  # Use enum values for JSON serialization
        arbitrary_types_allowed=True,  # Allow datetime and other complex types
    )

CloudSyncResult

Bases: BaseAgentCoreModel

Result of a cloud sync operation.

Source code in agentcore_cli/models/responses.py

class CloudSyncResult(BaseAgentCoreModel):
    """Result of a cloud sync operation."""

    success: bool = Field(description="Whether the operation was successful")
    message: str = Field(description="Status message")
    environment: str = Field(description="Environment name")
    synced_items: dict[str, int] = Field(default_factory=dict, description="Count of synced items by type")
    errors: list[str] = Field(default_factory=list, description="Errors encountered during sync")

CognitoConfig

Bases: ResourceBase

Cognito user pool and identity pool configuration.

Source code in agentcore_cli/models/resources.py

class CognitoConfig(ResourceBase):
    """Cognito user pool and identity pool configuration."""

    user_pool: CognitoUserPool | None = Field(default=None, description="User pool configuration")
    identity_pool: CognitoIdentityPool | None = Field(default=None, description="Identity pool configuration")
    last_sync: datetime | None = Field(default=None, description="Last sync timestamp")

CognitoIdentityPool

Bases: BaseAgentCoreModel

Cognito identity pool configuration.

Source code in agentcore_cli/models/resources.py

class CognitoIdentityPool(BaseAgentCoreModel):
    """Cognito identity pool configuration."""

    identity_pool_id: str = Field(description="Identity pool ID")
    identity_pool_name: str = Field(description="Identity pool name")
    created_at: datetime | None = Field(default=None, description="Creation timestamp")
    allow_unauthenticated_identities: bool = Field(
        default=False, description="Whether unauthenticated identities are allowed"
    )

CognitoUserPool

Bases: BaseAgentCoreModel

Cognito user pool configuration.

Source code in agentcore_cli/models/resources.py

class CognitoUserPool(BaseAgentCoreModel):
    """Cognito user pool configuration."""

    user_pool_id: str = Field(description="User pool ID")
    user_pool_name: str = Field(description="User pool name")
    user_pool_arn: str | None = Field(default=None, description="User pool ARN")
    client_id: str | None = Field(default=None, description="App client ID")
    client_secret: str | None = Field(default=None, description="App client secret")
    created_at: datetime | None = Field(default=None, description="Creation timestamp")
    domain: str | None = Field(default=None, description="User pool domain")

ContainerBuildInput

Bases: BaseAgentCoreModel

Input for building and pushing a container image to ECR.

Source code in agentcore_cli/models/inputs.py

class ContainerBuildInput(BaseAgentCoreModel):
    """Input for building and pushing a container image to ECR."""

    ecr_repository_name: str = Field(description="ECR repository name (must exist in configuration)")
    image_tag: str = Field(default="latest", description="Image tag to assign")
    dockerfile_path: str = Field(default="Dockerfile", description="Path to Dockerfile")
    build_context: str = Field(default=".", description="Docker build context directory")
    build_args: dict[str, str] = Field(default_factory=dict, description="Docker build arguments")
    platform: str = Field(default="linux/arm64", description="Target platform for the build")
    no_cache: bool = Field(default=False, description="Disable Docker build cache")

    @property
    def dockerfile(self) -> str:
        """Legacy property for backward compatibility."""
        return self.dockerfile_path

dockerfile property

Legacy property for backward compatibility.

CreateAgentRuntimeInput

Bases: BaseAgentCoreModel

Input for creating a new agent runtime.

Maps directly to AWS create_agent_runtime API parameters.

Source code in agentcore_cli/models/inputs.py

class CreateAgentRuntimeInput(BaseAgentCoreModel):
    """Input for creating a new agent runtime.

    Maps directly to AWS create_agent_runtime API parameters.
    """

    name: str = Field(description="Agent runtime name")
    container_uri: str = Field(description="Full ECR container URI including tag")
    role_arn: str = Field(description="IAM execution role ARN for the agent runtime")

    # Optional parameters
    description: str | None = Field(default=None, description="Agent runtime description")
    network_mode: NetworkModeType = Field(default=NetworkModeType.PUBLIC, description="Network configuration mode")
    protocol: ServerProtocolType = Field(default=ServerProtocolType.HTTP, description="Server protocol type")
    environment_variables: dict[str, str] = Field(default_factory=dict, description="Runtime environment variables")
    client_token: str | None = Field(default=None, description="Client token for idempotency")

CreateEndpointInput

Bases: BaseAgentCoreModel

Input for creating an agent runtime endpoint.

Maps to AWS create_agent_runtime_endpoint API parameters.

Source code in agentcore_cli/models/inputs.py

class CreateEndpointInput(BaseAgentCoreModel):
    """Input for creating an agent runtime endpoint.

    Maps to AWS create_agent_runtime_endpoint API parameters.
    """

    agent_runtime_id: str = Field(description="Agent runtime ID")
    name: str = Field(description="Endpoint name")
    target_version: str | None = Field(default=None, description="Specific runtime version (defaults to latest)")
    description: str | None = Field(default=None, description="Endpoint description")
    client_token: str | None = Field(default=None, description="Client token for idempotency")

CustomJWTAuthorizer

Bases: BaseAgentCoreModel

JWT authorizer configuration for AgentCore runtime.

Source code in agentcore_cli/models/runtime.py

class CustomJWTAuthorizer(BaseAgentCoreModel):
    """JWT authorizer configuration for AgentCore runtime."""

    discovery_url: str = Field(description="Discovery URL for JWT validation")
    allowed_audience: list[str] = Field(default_factory=list, description="Allowed audiences")
    allowed_clients: list[str] = Field(default_factory=list, description="Allowed clients")

ECRRepository

Bases: ResourceBase

ECR repository configuration.

Represents an Amazon ECR repository that can contain multiple image tags. Each AgentRuntimeVersion should reference this repository and specify an image tag.

Source code in agentcore_cli/models/resources.py

class ECRRepository(ResourceBase):
    """ECR repository configuration.

    Represents an Amazon ECR repository that can contain multiple image tags.
    Each AgentRuntimeVersion should reference this repository and specify an image tag.
    """

    name: str = Field(description="Repository name (e.g., 'my-chat-agent', 'my-data-processor')")
    registry_id: str = Field(description="AWS account ID that owns this repository")
    repository_uri: str = Field(
        description="Full repository URI without image tag (e.g., '123456789.dkr.ecr.us-east-1.amazonaws.com/my-agent')"
    )

    # Repository settings
    image_scanning_config: bool = Field(default=True, description="Whether images are scanned on push")
    image_tag_mutability: str = Field(default="MUTABLE", description="Image tag mutability: MUTABLE or IMMUTABLE")
    lifecycle_policy: dict[str, Any] | None = Field(default=None, description="Repository lifecycle policy")

    # Image tracking
    available_tags: set[str] = Field(default_factory=set, description="Set of available image tags in this repository")
    last_push: datetime | None = Field(default=None, description="Last image push timestamp")
    last_sync: datetime | None = Field(default=None, description="Last sync timestamp with AWS")

    @property
    def registry_url(self) -> str:
        """Get the ECR registry URL (without repository name)."""
        # Extract registry from repository_uri
        # Format: registry_id.dkr.ecr.region.amazonaws.com/repo_name
        if "/" in self.repository_uri:
            return self.repository_uri.split("/")[0]
        return self.repository_uri

    def get_image_uri(self, tag: str) -> str:
        """Get the full container image URI for a specific tag.

        Args:
            tag: Image tag (e.g., 'v1', 'latest', 'prod-2024-01-15')

        Returns:
            Full container URI: registry/repository:tag
        """
        return f"{self.repository_uri}:{tag}"

    def validate_tag_exists(self, tag: str) -> bool:
        """Check if an image tag exists in this repository."""
        return tag in self.available_tags

    @field_validator("repository_uri")
    @classmethod
    def validate_repository_uri(cls, v: str) -> str:
        """Validate ECR repository URI format."""
        if not v:
            raise ValueError("Repository URI cannot be empty")

        # Should match ECR pattern: account.dkr.ecr.region.amazonaws.com/repo-name
        import re

        ecr_pattern = r"^[0-9]+\.dkr\.ecr\.[a-z0-9-]+\.amazonaws\.com\/[a-z0-9][a-z0-9._-]*$"
        if not re.match(ecr_pattern, v):
            raise ValueError(
                f"Repository URI '{v}' does not match ECR format: account.dkr.ecr.region.amazonaws.com/repository-name"
            )

        return v

registry_url property

Get the ECR registry URL (without repository name).

get_image_uri(tag)

Get the full container image URI for a specific tag.

Parameters:

Name	Type	Description	Default
tag	str	Image tag (e.g., 'v1', 'latest', 'prod-2024-01-15')	required

Returns:

Type	Description
str	Full container URI: registry/repository:tag

Source code in agentcore_cli/models/resources.py

def get_image_uri(self, tag: str) -> str:
    """Get the full container image URI for a specific tag.

    Args:
        tag: Image tag (e.g., 'v1', 'latest', 'prod-2024-01-15')

    Returns:
        Full container URI: registry/repository:tag
    """
    return f"{self.repository_uri}:{tag}"

validate_repository_uri(v) classmethod

Validate ECR repository URI format.

Source code in agentcore_cli/models/resources.py

@field_validator("repository_uri")
@classmethod
def validate_repository_uri(cls, v: str) -> str:
    """Validate ECR repository URI format."""
    if not v:
        raise ValueError("Repository URI cannot be empty")

    # Should match ECR pattern: account.dkr.ecr.region.amazonaws.com/repo-name
    import re

    ecr_pattern = r"^[0-9]+\.dkr\.ecr\.[a-z0-9-]+\.amazonaws\.com\/[a-z0-9][a-z0-9._-]*$"
    if not re.match(ecr_pattern, v):
        raise ValueError(
            f"Repository URI '{v}' does not match ECR format: account.dkr.ecr.region.amazonaws.com/repository-name"
        )

    return v

validate_tag_exists(tag)

Check if an image tag exists in this repository.

Source code in agentcore_cli/models/resources.py

def validate_tag_exists(self, tag: str) -> bool:
    """Check if an image tag exists in this repository."""
    return tag in self.available_tags

EndpointCreationResult

Bases: ActionResult

Result of endpoint creation.

Source code in agentcore_cli/models/responses.py

class EndpointCreationResult(ActionResult):
    """Result of endpoint creation."""

    agent_name: str = Field(description="Agent name")
    runtime_id: str = Field(description="Runtime ID")
    endpoint_name: str = Field(description="Endpoint name")
    endpoint_arn: str | None = Field(default=None, description="Endpoint ARN")
    target_version: str = Field(description="Version the endpoint points to")
    environment: str | None = Field(default=None, description="Environment mapped to this endpoint")

EndpointUpdateResult

Bases: ActionResult

Result of endpoint update.

Source code in agentcore_cli/models/responses.py

class EndpointUpdateResult(ActionResult):
    """Result of endpoint update."""

    agent_name: str = Field(description="Agent name")
    runtime_id: str = Field(description="Runtime ID")
    endpoint_name: str = Field(description="Endpoint name")
    previous_version: str = Field(description="Previous version")
    new_version: str = Field(description="New version the endpoint points to")
    environment: str | None = Field(default=None, description="Environment mapped to this endpoint")

EnvironmentConfig

Bases: BaseAgentCoreModel

Configuration for a specific environment (dev, staging, prod).

Each environment owns its agent runtimes and exists in a specific AWS region. This design aligns with AWS AgentCore where runtimes are region-specific resources.

Source code in agentcore_cli/models/config.py

class EnvironmentConfig(BaseAgentCoreModel):
    """Configuration for a specific environment (dev, staging, prod).

    Each environment owns its agent runtimes and exists in a specific AWS region.
    This design aligns with AWS AgentCore where runtimes are region-specific resources.
    """

    name: str = Field(description="Environment name (e.g., 'dev', 'staging', 'prod')")
    region: str = Field(description="AWS region for this environment and all its agent runtimes")

    # Agent runtimes owned by this environment
    agent_runtimes: dict[str, "AgentRuntime"] = Field(
        default_factory=dict, description="Agent runtimes deployed in this environment, keyed by runtime name"
    )

    # Default agent for this environment
    default_agent_runtime: str | None = Field(
        default=None, description="Default agent runtime name to use for operations (must exist in agent_runtimes)"
    )

    # Environment-specific settings
    environment_variables: dict[str, str] = Field(
        default_factory=dict, description="Default environment variables for all runtimes in this environment"
    )

    # Auth configuration for this environment
    cognito: CognitoConfig | None = Field(default=None, description="Cognito configuration for this environment")

    # Metadata
    created_at: datetime | None = Field(default=None, description="When this environment was created")
    updated_at: datetime | None = Field(default=None, description="When this environment was last updated")

    @model_validator(mode="after")
    def set_creation_time(cls, model):
        """Set creation time if not set."""
        if model.created_at is None:
            model.created_at = datetime.now()
        return model

    @model_validator(mode="after")
    def validate_default_agent(cls, model):
        """Validate that default_agent_runtime exists in agent_runtimes."""
        if model.default_agent_runtime:
            if model.default_agent_runtime not in model.agent_runtimes:
                available_runtimes = list(model.agent_runtimes.keys())
                raise ValueError(
                    f"default_agent_runtime '{model.default_agent_runtime}' does not exist in agent_runtimes. "
                    f"Available runtimes: {available_runtimes}"
                )
        return model

    @model_validator(mode="after")
    def validate_runtime_regions(cls, model):
        """Ensure all agent runtimes in this environment match the environment's region."""
        for runtime_name, runtime in model.agent_runtimes.items():
            if hasattr(runtime, "region") and runtime.region != model.region:
                raise ValueError(
                    f"Agent runtime '{runtime_name}' is in region '{runtime.region}' "
                    f"but environment '{model.name}' is in region '{model.region}'. "
                    f"All runtimes in an environment must be in the same region."
                )
        return model

    def get_agent_endpoint(self, agent_runtime_name: str, endpoint_name: str | None = None) -> tuple[str, str] | None:
        """Get the agent runtime and endpoint for invocation.

        Args:
            agent_runtime_name: Name of the agent runtime
            endpoint_name: Specific endpoint name, or None to use DEFAULT

        Returns:
            Tuple of (runtime_name, endpoint_name) or None if not found
        """
        if agent_runtime_name not in self.agent_runtimes:
            return None

        runtime = self.agent_runtimes[agent_runtime_name]
        target_endpoint = endpoint_name or "DEFAULT"

        if target_endpoint not in runtime.endpoints:
            return None

        return (agent_runtime_name, target_endpoint)

get_agent_endpoint(agent_runtime_name, endpoint_name=None)

Get the agent runtime and endpoint for invocation.

Parameters:

Name	Type	Description	Default
agent_runtime_name	str	Name of the agent runtime	required
endpoint_name	str | None	Specific endpoint name, or None to use DEFAULT	None

Returns:

Type	Description
tuple[str, str] | None	Tuple of (runtime_name, endpoint_name) or None if not found

Source code in agentcore_cli/models/config.py

def get_agent_endpoint(self, agent_runtime_name: str, endpoint_name: str | None = None) -> tuple[str, str] | None:
    """Get the agent runtime and endpoint for invocation.

    Args:
        agent_runtime_name: Name of the agent runtime
        endpoint_name: Specific endpoint name, or None to use DEFAULT

    Returns:
        Tuple of (runtime_name, endpoint_name) or None if not found
    """
    if agent_runtime_name not in self.agent_runtimes:
        return None

    runtime = self.agent_runtimes[agent_runtime_name]
    target_endpoint = endpoint_name or "DEFAULT"

    if target_endpoint not in runtime.endpoints:
        return None

    return (agent_runtime_name, target_endpoint)

set_creation_time(model)

Set creation time if not set.

Source code in agentcore_cli/models/config.py

@model_validator(mode="after")
def set_creation_time(cls, model):
    """Set creation time if not set."""
    if model.created_at is None:
        model.created_at = datetime.now()
    return model

validate_default_agent(model)

Validate that default_agent_runtime exists in agent_runtimes.

Source code in agentcore_cli/models/config.py

@model_validator(mode="after")
def validate_default_agent(cls, model):
    """Validate that default_agent_runtime exists in agent_runtimes."""
    if model.default_agent_runtime:
        if model.default_agent_runtime not in model.agent_runtimes:
            available_runtimes = list(model.agent_runtimes.keys())
            raise ValueError(
                f"default_agent_runtime '{model.default_agent_runtime}' does not exist in agent_runtimes. "
                f"Available runtimes: {available_runtimes}"
            )
    return model

validate_runtime_regions(model)

Ensure all agent runtimes in this environment match the environment's region.

Source code in agentcore_cli/models/config.py

@model_validator(mode="after")
def validate_runtime_regions(cls, model):
    """Ensure all agent runtimes in this environment match the environment's region."""
    for runtime_name, runtime in model.agent_runtimes.items():
        if hasattr(runtime, "region") and runtime.region != model.region:
            raise ValueError(
                f"Agent runtime '{runtime_name}' is in region '{runtime.region}' "
                f"but environment '{model.name}' is in region '{model.region}'. "
                f"All runtimes in an environment must be in the same region."
            )
    return model

EnvironmentCreationResult

Bases: ActionResult

Result of environment creation.

Source code in agentcore_cli/models/responses.py

class EnvironmentCreationResult(ActionResult):
    """Result of environment creation."""

    name: str = Field(description="Environment name")
    region: str = Field(description="AWS region")

EnvironmentDeletionResult

Bases: ActionResult

Result of environment deletion.

Source code in agentcore_cli/models/responses.py

class EnvironmentDeletionResult(ActionResult):
    """Result of environment deletion."""

    name: str = Field(description="Environment name")

EnvironmentInput

Bases: BaseAgentCoreModel

Input for environment operations.

Source code in agentcore_cli/models/inputs.py

class EnvironmentInput(BaseAgentCoreModel):
    """Input for environment operations."""

    name: str = Field(description="Environment name")
    region: str | None = Field(default=None, description="AWS region for the environment")

GlobalResourceConfig

Bases: BaseAgentCoreModel

Global resources shared across environments.

Source code in agentcore_cli/models/config.py

class GlobalResourceConfig(BaseAgentCoreModel):
    """Global resources shared across environments."""

    # ECR repositories (can be shared across environments)
    ecr_repositories: dict[str, "ECRRepository"] = Field(default_factory=dict, description="ECR repositories by name")

    # IAM roles (can be shared across environments)
    iam_roles: dict[str, "IAMRoleConfig"] = Field(default_factory=dict, description="IAM roles by name")

    # Sync configuration
    sync_config: SyncConfig = Field(default_factory=SyncConfig, description="Sync configuration")

IAMRoleConfig

Bases: ResourceBase

IAM role configuration.

Source code in agentcore_cli/models/resources.py

class IAMRoleConfig(ResourceBase):
    """IAM role configuration."""

    name: str = Field(description="Role name")
    arn: str = Field(description="Role ARN")
    path: str = Field(default="/", description="Role path")
    description: str | None = Field(default=None, description="Role description")
    assume_role_policy_document: dict[str, Any] | None = Field(default=None, description="Assume role policy")
    last_sync: datetime | None = Field(default=None, description="Last sync timestamp with AWS")

ImageBuildResult

Bases: ActionResult

Result of image build.

Source code in agentcore_cli/models/responses.py

class ImageBuildResult(ActionResult):
    """Result of image build."""

    repo_name: str = Field(description="Repository name")
    tag: str = Field(description="Image tag")
    image_id: str | None = Field(default=None, description="Image ID")

InvokeAgentInput

Bases: BaseAgentCoreModel

Input for invoking an agent runtime.

Maps directly to AWS invoke_agent_runtime API requirements. The API requires a full ARN and qualifier (endpoint name).

Source code in agentcore_cli/models/inputs.py

class InvokeAgentInput(BaseAgentCoreModel):
    """Input for invoking an agent runtime.

    Maps directly to AWS invoke_agent_runtime API requirements.
    The API requires a full ARN and qualifier (endpoint name).
    """

    agent_runtime_arn: str = Field(description="Full ARN of the agent runtime to invoke")
    qualifier: str = Field(description="Endpoint name or version qualifier (e.g., 'DEFAULT', 'production')")
    runtime_session_id: str = Field(description="Session ID for the runtime invocation")
    prompt: str = Field(description="Prompt to send to the agent")

    # Additional fields for CLI convenience
    environment: str | None = Field(default=None, description="Environment context (for CLI reference only)")
    agent_name: str | None = Field(default=None, description="Agent name (for CLI reference only)")

NetworkModeType

Bases: StrEnum

Network mode for AgentCore runtimes.

Defines how the agent runtime is exposed within AWS networking: - PUBLIC: Runtime is accessible from the internet via public endpoints - PRIVATE: Runtime is only accessible within VPC (future AWS enhancement)

Source code in agentcore_cli/models/base.py

class NetworkModeType(StrEnum):
    """Network mode for AgentCore runtimes.

    Defines how the agent runtime is exposed within AWS networking:
    - PUBLIC: Runtime is accessible from the internet via public endpoints
    - PRIVATE: Runtime is only accessible within VPC (future AWS enhancement)
    """

    PUBLIC = "PUBLIC"

ResourceBase

Bases: BaseAgentCoreModel

Base class for all AWS resource models.

Provides common fields that all AWS resources share: region, timestamps, and tags for resource management.

Source code in agentcore_cli/models/base.py

class ResourceBase(BaseAgentCoreModel):
    """Base class for all AWS resource models.

    Provides common fields that all AWS resources share:
    region, timestamps, and tags for resource management.
    """

    region: str = Field(description="AWS region where this resource exists")
    created_at: datetime | None = Field(default=None, description="Resource creation timestamp")
    updated_at: datetime | None = Field(default=None, description="Last modification timestamp")
    tags: dict[str, str] = Field(default_factory=dict, description="AWS resource tags")

ResourceTag

Bases: BaseAgentCoreModel

AWS resource tag for cost allocation and resource management.

Source code in agentcore_cli/models/base.py

class ResourceTag(BaseAgentCoreModel):
    """AWS resource tag for cost allocation and resource management."""

    key: str = Field(description="Tag key (case-sensitive)")
    value: str = Field(description="Tag value")

ServerProtocolType

Bases: StrEnum

Server protocol type for AgentCore runtimes.

Defines the communication protocol the agent runtime uses: - HTTP: Standard HTTP protocol for REST API communication - MCP: Model Context Protocol for advanced agent interactions

Source code in agentcore_cli/models/base.py

class ServerProtocolType(StrEnum):
    """Server protocol type for AgentCore runtimes.

    Defines the communication protocol the agent runtime uses:
    - HTTP: Standard HTTP protocol for REST API communication
    - MCP: Model Context Protocol for advanced agent interactions
    """

    HTTP = "HTTP"
    MCP = "MCP"

SyncConfig

Bases: BaseAgentCoreModel

Configuration for cloud sync behavior.

Source code in agentcore_cli/models/config.py

class SyncConfig(BaseAgentCoreModel):
    """Configuration for cloud sync behavior."""

    cloud_config_enabled: bool = Field(default=False, description="Whether cloud config is enabled")
    auto_sync_enabled: bool = Field(default=True, description="Whether auto-sync is enabled")
    parameter_store_prefix: str = Field(default="/agentcore", description="Parameter Store prefix")
    last_full_sync: datetime | None = Field(default=None, description="Last full sync timestamp")
    sync_interval_minutes: int = Field(default=60, description="Sync interval in minutes when auto-sync is enabled")

SyncStatus

Bases: BaseAgentCoreModel

Status of configuration synchronization.

Source code in agentcore_cli/models/responses.py

class SyncStatus(BaseAgentCoreModel):
    """Status of configuration synchronization."""

    environment: str = Field(description="Environment name")
    cloud_config_enabled: bool = Field(description="Whether cloud config is enabled")
    auto_sync_enabled: bool = Field(description="Whether auto-sync is enabled")
    last_sync: datetime | None = Field(default=None, description="Last sync timestamp")
    in_sync: bool = Field(description="Whether local and cloud are in sync")
    drift_details: dict[str, dict[str, list[str]]] | None = Field(default=None, description="Drift details")

UpdateAgentRuntimeInput

Bases: BaseAgentCoreModel

Input for updating an agent runtime.

Updates create a new immutable version. Maps to AWS update_agent_runtime API.

Source code in agentcore_cli/models/inputs.py

class UpdateAgentRuntimeInput(BaseAgentCoreModel):
    """Input for updating an agent runtime.

    Updates create a new immutable version. Maps to AWS update_agent_runtime API.
    """

    agent_runtime_id: str = Field(description="Agent runtime ID to update")

    # Optional update parameters - at least one must be provided
    description: str | None = Field(default=None, description="Updated description")
    container_uri: str | None = Field(default=None, description="New container URI")
    role_arn: str | None = Field(default=None, description="New IAM execution role ARN")
    network_mode: NetworkModeType | None = Field(default=None, description="New network mode")
    protocol: ServerProtocolType | None = Field(default=None, description="New protocol type")
    environment_variables: dict[str, str] | None = Field(default=None, description="Updated environment variables")
    client_token: str | None = Field(default=None, description="Client token for idempotency")

    @model_validator(mode="after")
    def validate_at_least_one_update(cls, model):
        """Ensure at least one field is being updated."""
        update_fields = [
            model.description,
            model.container_uri,
            model.role_arn,
            model.network_mode,
            model.protocol,
            model.environment_variables,
        ]
        if not any(field is not None for field in update_fields):
            raise ValueError("At least one field must be provided for update")
        return model

validate_at_least_one_update(model)

Ensure at least one field is being updated.

Source code in agentcore_cli/models/inputs.py

@model_validator(mode="after")
def validate_at_least_one_update(cls, model):
    """Ensure at least one field is being updated."""
    update_fields = [
        model.description,
        model.container_uri,
        model.role_arn,
        model.network_mode,
        model.protocol,
        model.environment_variables,
    ]
    if not any(field is not None for field in update_fields):
        raise ValueError("At least one field must be provided for update")
    return model

UpdateEndpointInput

Bases: BaseAgentCoreModel

Input for updating an agent runtime endpoint.

Maps to AWS update_agent_runtime_endpoint API parameters.

Source code in agentcore_cli/models/inputs.py

class UpdateEndpointInput(BaseAgentCoreModel):
    """Input for updating an agent runtime endpoint.

    Maps to AWS update_agent_runtime_endpoint API parameters.
    """

    agent_runtime_id: str = Field(description="Agent runtime ID")
    endpoint_name: str = Field(description="Endpoint name to update")
    target_version: str = Field(description="New runtime version to point endpoint to")
    description: str | None = Field(default=None, description="Updated description")
    client_token: str | None = Field(default=None, description="Client token for idempotency")

WorkloadIdentity

Bases: BaseAgentCoreModel

Workload identity details for AgentCore runtime.

Source code in agentcore_cli/models/runtime.py

class WorkloadIdentity(BaseAgentCoreModel):
    """Workload identity details for AgentCore runtime."""

    workload_identity_arn: str = Field(description="Workload identity ARN")