invoke_agent_runtime ¶

@contextmanager
def invoke_agent_runtime(
    self,
    agent_runtime_arn: str,
    payload: "capo_bedrock_agentcore.types.body.Body",
    *,
    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,
    mcp_session_id: Optional[
        "capo_bedrock_agentcore.types.string_type.StringType"
    ] = None,
    runtime_session_id: Optional[
        "capo_bedrock_agentcore.types.session_type.SessionType"
    ] = None,
    mcp_protocol_version: Optional[
        "capo_bedrock_agentcore.types.string_type.StringType"
    ] = None,
    mcp_method: Optional[
        "capo_bedrock_agentcore.types.string_type.StringType"
    ] = None,
    mcp_name: Optional[
        "capo_bedrock_agentcore.types.string_type.StringType"
    ] = None,
    runtime_user_id: Optional[
        "capo_bedrock_agentcore.types.string_type.StringType"
    ] = 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_response.InvokeAgentRuntimeResponse]":
    """<p>Sends a request to an agent or tool hosted in an Amazon Bedrock AgentCore Runtime and receives responses in real-time. </p> <p>To invoke an agent, you can specify either the AgentCore Runtime ARN or the agent ID with an account ID, and provide a payload containing your request. When you use the agent ID instead of the full ARN, you don't need to URL-encode the identifier. You can optionally specify a qualifier to target a specific endpoint of the agent.</p> <p>This operation supports streaming responses, allowing you to receive partial responses as they become available. We recommend using pagination to ensure that the operation returns quickly and successfully when processing large responses.</p> <p>For example code, see <a href="https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-invoke-agent.html">Invoke an AgentCore Runtime agent</a>. </p> <p>If you're integrating your agent with OAuth, you can't use the Amazon Web Services SDK to call <code>InvokeAgentRuntime</code>. Instead, make a HTTPS request to <code>InvokeAgentRuntime</code>. For an example, see <a href="https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-oauth.html">Authenticate and authorize with Inbound Auth and Outbound Auth</a>.</p> <p>To use this operation, you must have the <code>bedrock-agentcore:InvokeAgentRuntime</code> permission. If you are making a call to <code>InvokeAgentRuntime</code> on behalf of a user ID with the <code>X-Amzn-Bedrock-AgentCore-Runtime-User-Id</code> header, You require permissions to both actions (<code>bedrock-agentcore:InvokeAgentRuntime</code> and <code>bedrock-agentcore:InvokeAgentRuntimeForUser</code>). </p>

    Args:
        content_type: <p>The MIME type of the input data in the 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. This tells the agent runtime what format to use for the response data. Common values include application/json for JSON data.</p>
        mcp_session_id: <p>The identifier of the MCP session.</p>
        runtime_session_id: <p>The identifier of the runtime session.</p>
        mcp_protocol_version: <p>The version of the MCP protocol being used.</p>
        mcp_method: <p>The MCP method being invoked. For example, <code>tools/call</code>, <code>resources/read</code>, or <code>prompts/get</code>.</p>
        mcp_name: <p>The name of the MCP resource, tool, or prompt being accessed. The value depends on the method:</p> <ul> <li> <p> <code>tools/call</code> – The tool name.</p> </li> <li> <p> <code>resources/read</code> – The resource URI.</p> </li> <li> <p> <code>prompts/get</code> – The prompt name.</p> </li> </ul>
        runtime_user_id: <p>The identifier of the runtime user.</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 identifier of the agent runtime to invoke. You can specify either the full Amazon Web Services Resource Name (ARN) or the agent ID. If you use the agent ID, you must also provide the <code>accountId</code> query parameter.</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>
        payload: <p>The input data to send to the agent runtime. The format of this data depends on the specific agent configuration and must match the specified content type. For most agents, this is a JSON object containing the user's request.</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_request.InvokeAgentRuntimeRequest]",
    ) -> OperationResponse[
        "capo_bedrock_agentcore.types.invoke_agent_runtime_response.InvokeAgentRuntimeResponse"
    ]:
        import capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.invoke_agent_runtime

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

    def _redact_input(
        input_: "capo_bedrock_agentcore.types.invoke_agent_runtime_request.InvokeAgentRuntimeRequest",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore.types.invoke_agent_runtime_request

        return capo_bedrock_agentcore.types.invoke_agent_runtime_request.redact_sensitive(
            input_
        )

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

        return capo_bedrock_agentcore.types.invoke_agent_runtime_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "InvokeAgentRuntime",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_agentcore.types.invoke_agent_runtime_request.InvokeAgentRuntimeRequest = {
        "agent_runtime_arn": agent_runtime_arn,
        "payload": payload,
    }
    if content_type is not None:
        input_["content_type"] = content_type
    if accept is not None:
        input_["accept"] = accept
    if mcp_session_id is not None:
        input_["mcp_session_id"] = mcp_session_id
    if runtime_session_id is None:
        runtime_session_id = str(uuid.uuid4())
    input_["runtime_session_id"] = runtime_session_id
    if mcp_protocol_version is not None:
        input_["mcp_protocol_version"] = mcp_protocol_version
    if mcp_method is not None:
        input_["mcp_method"] = mcp_method
    if mcp_name is not None:
        input_["mcp_name"] = mcp_name
    if runtime_user_id is not None:
        input_["runtime_user_id"] = runtime_user_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()