Skip to content

agentcore_cli.models.responses

agentcore_cli.models.responses

Response models for AgentCore Platform CLI.

This module defines models for responses from AgentCore operations and CLI commands.

ActionResult

Bases: BaseAgentCoreModel

Base model for CLI action results.

Source code in agentcore_cli/models/responses.py
Python
122
123
124
125
126
class ActionResult(BaseAgentCoreModel):
    """Base model for CLI action results."""

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

AgentCreationResult

Bases: ActionResult

Result of agent creation.

Source code in agentcore_cli/models/responses.py
Python
129
130
131
132
133
134
135
136
137
138
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
Python
173
174
175
176
177
178
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")

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
Python
13
14
15
16
17
18
19
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
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
Python
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
@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
Python
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
@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,
    )

AgentUpdateResult

Bases: ActionResult

Result of agent update.

Source code in agentcore_cli/models/responses.py
Python
141
142
143
144
145
146
147
148
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")

CloudSyncResult

Bases: BaseAgentCoreModel

Result of a cloud sync operation.

Source code in agentcore_cli/models/responses.py
Python
112
113
114
115
116
117
118
119
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")

EndpointCreationResult

Bases: ActionResult

Result of endpoint creation.

Source code in agentcore_cli/models/responses.py
Python
151
152
153
154
155
156
157
158
159
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
Python
162
163
164
165
166
167
168
169
170
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")

EnvironmentCreationResult

Bases: ActionResult

Result of environment creation.

Source code in agentcore_cli/models/responses.py
Python
181
182
183
184
185
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
Python
188
189
190
191
class EnvironmentDeletionResult(ActionResult):
    """Result of environment deletion."""

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

ImageBuildResult

Bases: ActionResult

Result of image build.

Source code in agentcore_cli/models/responses.py
Python
194
195
196
197
198
199
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")

SyncStatus

Bases: BaseAgentCoreModel

Status of configuration synchronization.

Source code in agentcore_cli/models/responses.py
Python
101
102
103
104
105
106
107
108
109
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")