invoke_agent_runtime_command ¶

@contextmanager
def invoke_agent_runtime_command(
    self,
    agent_runtime_arn: str,
    body: "capo_bedrock_agentcore.types.invoke_agent_runtime_command_request_body.InvokeAgentRuntimeCommandRequestBody",
    *,
    config_overrides: Optional[BedrockAgentCoreClientConfig] = None,
    content_type: Optional[
        "capo_bedrock_agentcore.types.mime_type.MimeType"
    ] = None,
    accept: Optional["capo_bedrock_agentcore.types.mime_type.MimeType"] = None,
    runtime_session_id: Optional[
        "capo_bedrock_agentcore.types.session_type.SessionType"
    ] = None,
    trace_id: Optional[str] = None,
    trace_parent: Optional[str] = None,
    trace_state: Optional[str] = None,
    baggage: Optional[str] = None,
    qualifier: Optional[str] = None,
    account_id: Optional[str] = None,
) -> "Generator[capo_bedrock_agentcore.types.invoke_agent_runtime_command_response.InvokeAgentRuntimeCommandResponse]":
    """<p>Executes a command in a runtime session container and streams the output back to the caller. This operation allows you to run shell commands within the agent runtime environment and receive real-time streaming responses including standard output and standard error.</p> <p>To invoke a command, you must specify the agent runtime ARN and a runtime session ID. The command execution supports streaming responses, allowing you to receive output as it becomes available through <code>contentStart</code>, <code>contentDelta</code>, and <code>contentStop</code> events.</p> <p>To use this operation, you must have the <code>bedrock-agentcore:InvokeAgentRuntimeCommand</code> permission.</p>

    Args:
        content_type: <p>The MIME type of the input data in the request payload. This tells the agent runtime how to interpret the payload data. Common values include application/json for JSON data.</p>
        accept: <p>The desired MIME type for the response from the agent runtime command. This tells the agent runtime what format to use for the response data. Common values include application/json for JSON data.</p>
        runtime_session_id: <p>The unique identifier of the runtime session in which to execute the command. This session ID is used to maintain state and context across multiple command invocations.</p>
        trace_id: <p>The trace identifier for request tracking.</p>
        trace_parent: <p>The parent trace information for distributed tracing.</p>
        trace_state: <p>The trace state information for distributed tracing.</p>
        baggage: <p>Additional context information for distributed tracing.</p>
        agent_runtime_arn: <p>The Amazon Resource Name (ARN) of the agent runtime on which to execute the command. This identifies the specific agent runtime environment where the command will run.</p>
        qualifier: <p>The qualifier to use for the agent runtime. This is an endpoint name that points to a specific version. If not specified, Amazon Bedrock AgentCore uses the default endpoint of the agent runtime.</p>
        account_id: <p>The identifier of the Amazon Web Services account for the agent runtime resource. This parameter is required when you specify an agent ID instead of the full ARN for <code>agentRuntimeArn</code>.</p>
        body: <p>The request body containing the command to execute and optional configuration parameters such as timeout settings.</p>

    Raises:
        capo_bedrock_agentcore.errors.access_denied_exception.AccessDeniedException: <p>The exception that occurs when you do not have sufficient permissions to perform an action. Verify that your IAM policy includes the necessary permissions for the operation you are trying to perform.</p>
        capo_bedrock_agentcore.errors.internal_server_exception.InternalServerException: <p>The exception that occurs when the service encounters an unexpected internal error. This is a temporary condition that will resolve itself with retries. We recommend implementing exponential backoff retry logic in your application.</p>
        capo_bedrock_agentcore.errors.resource_not_found_exception.ResourceNotFoundException: <p>The exception that occurs when the specified resource does not exist. This can happen when using an invalid identifier or when trying to access a resource that has been deleted.</p>
        capo_bedrock_agentcore.errors.retryable_conflict_exception.RetryableConflictException: <p>The exception that occurs when there is a retryable conflict performing an operation. This is a temporary condition that may resolve itself with retries. We recommend implementing exponential backoff retry logic in your application.</p>
        capo_bedrock_agentcore.errors.runtime_client_error.RuntimeClientError: <p>The exception that occurs when there is an error in the runtime client. This can happen due to network issues, invalid configuration, or other client-side problems. Check the error message for specific details about the error.</p>
        capo_bedrock_agentcore.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: <p>The exception that occurs when the request would cause a service quota to be exceeded. Review your service quotas and either reduce your request rate or request a quota increase.</p>
        capo_bedrock_agentcore.errors.throttling_exception.ThrottlingException: <p>The exception that occurs when the request was denied due to request throttling. This happens when you exceed the allowed request rate for an operation. Reduce the frequency of requests or implement exponential backoff retry logic in your application.</p>
        capo_bedrock_agentcore.errors.validation_exception.ValidationException: <p>The exception that occurs when the input fails to satisfy the constraints specified by the service. Check the error message for details about which input parameter is invalid and correct your request.</p>
        capo_bedrock_agentcore.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_bedrock_agentcore.types.invoke_agent_runtime_command_request.InvokeAgentRuntimeCommandRequest]",
    ) -> OperationResponse[
        "capo_bedrock_agentcore.types.invoke_agent_runtime_command_response.InvokeAgentRuntimeCommandResponse"
    ]:
        import capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.invoke_agent_runtime_command

        output, http_response = (
            capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.invoke_agent_runtime_command.invoke_agent_runtime_command(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_bedrock_agentcore.types.invoke_agent_runtime_command_response.InvokeAgentRuntimeCommandResponse",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore.types.invoke_agent_runtime_command_response

        return capo_bedrock_agentcore.types.invoke_agent_runtime_command_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("InvokeAgentRuntimeCommand", redact_output=_redact_output),
    )
    input_: capo_bedrock_agentcore.types.invoke_agent_runtime_command_request.InvokeAgentRuntimeCommandRequest = {
        "agent_runtime_arn": agent_runtime_arn,
        "body": body,
    }
    if content_type is not None:
        input_["content_type"] = content_type
    if accept is not None:
        input_["accept"] = accept
    if runtime_session_id is None:
        runtime_session_id = str(uuid.uuid4())
    input_["runtime_session_id"] = runtime_session_id
    if trace_id is not None:
        input_["trace_id"] = trace_id
    if trace_parent is not None:
        input_["trace_parent"] = trace_parent
    if trace_state is not None:
        input_["trace_state"] = trace_state
    if baggage is not None:
        input_["baggage"] = baggage
    if qualifier is not None:
        input_["qualifier"] = qualifier
    if account_id is not None:
        input_["account_id"] = account_id

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    try:
        yield response.output
    finally:
        response.response.close()