invoke_harness ¶

@contextmanager
def invoke_harness(
    self,
    harness_arn: "capo_bedrock_agentcore.types.harness_arn.HarnessArn",
    runtime_session_id: "capo_bedrock_agentcore.types.session_id.SessionId",
    messages: "capo_bedrock_agentcore.types.harness_messages.HarnessMessages",
    *,
    config_overrides: Optional[BedrockAgentCoreClientConfig] = None,
    qualifier: Optional[
        "capo_bedrock_agentcore.types.harness_endpoint_name.HarnessEndpointName"
    ] = None,
    runtime_user_id: Optional[str] = None,
    trace_parent: Optional[str] = None,
    trace_state: Optional[str] = None,
    trace_id: Optional[str] = None,
    baggage: Optional[str] = None,
    model: Optional[
        "capo_bedrock_agentcore.types.harness_model_configuration.HarnessModelConfiguration"
    ] = None,
    system_prompt: Optional[
        "capo_bedrock_agentcore.types.harness_system_prompt.HarnessSystemPrompt"
    ] = None,
    tools: Optional[
        "capo_bedrock_agentcore.types.harness_tools.HarnessTools"
    ] = None,
    skills: Optional[
        "capo_bedrock_agentcore.types.harness_skills.HarnessSkills"
    ] = None,
    allowed_tools: Optional[
        "capo_bedrock_agentcore.types.harness_allowed_tools.HarnessAllowedTools"
    ] = None,
    max_iterations: Optional[int] = None,
    max_tokens: Optional[int] = None,
    timeout_seconds: Optional[int] = None,
    actor_id: Optional[str] = None,
) -> "Generator[capo_bedrock_agentcore.types.invoke_harness_response.InvokeHarnessResponse]":
    """<p>Operation to invoke a Harness.</p>

    Args:
        harness_arn: <p>The ARN of the harness to invoke.</p>
        qualifier: <p>The endpoint name to invoke. If omitted, the DEFAULT endpoint is used.</p>
        runtime_session_id: <p>The session ID for the invocation. Use the same session ID across requests to continue a conversation.</p>
        runtime_user_id: <p>An identifier for the end user making the request. This value is passed through to the runtime container.</p>
        trace_parent: <p>W3C trace context parent header containing version, trace ID, parent span ID, and trace flags.</p>
        trace_state: <p>W3C trace context state header for vendor-specific trace information.</p>
        trace_id: <p>Trace ID for maintaining observability through the operation.</p>
        baggage: <p>W3C Baggage header for user-defined context propagation. Format: key1=value1,key2=value2</p>
        messages: <p>The messages to send to the agent.</p>
        model: <p>The model configuration to use for this invocation. If specified, overrides the harness default.</p>
        system_prompt: <p>The system prompt to use for this invocation. If specified, overrides the harness default.</p>
        tools: <p>The tools available to the agent for this invocation. If specified, overrides the harness default.</p>
        skills: <p>The skills available to the agent for this invocation. If specified, overrides the harness default.</p>
        allowed_tools: <p>The tools that the agent is allowed to use for this invocation. If specified, overrides the harness default.</p>
        max_iterations: <p>The maximum number of iterations the agent loop can execute. If specified, overrides the harness default.</p>
        max_tokens: <p>The maximum number of tokens the agent can generate per iteration. If specified, overrides the harness default.</p>
        timeout_seconds: <p>The maximum duration in seconds for the agent loop execution. If specified, overrides the harness default.</p>
        actor_id: <p>The actor ID for memory operations. Overrides the actor ID configured on the harness.</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.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_harness_request.InvokeHarnessRequest]",
    ) -> OperationResponse[
        "capo_bedrock_agentcore.types.invoke_harness_response.InvokeHarnessResponse"
    ]:
        import capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.invoke_harness

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

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

        return capo_bedrock_agentcore.types.invoke_harness_request.redact_sensitive(
            input_
        )

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

        return (
            capo_bedrock_agentcore.types.invoke_harness_response.redact_sensitive(
                output
            )
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "InvokeHarness",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_agentcore.types.invoke_harness_request.InvokeHarnessRequest = {
        "harness_arn": harness_arn,
        "runtime_session_id": runtime_session_id,
        "messages": messages,
    }
    if qualifier is not None:
        input_["qualifier"] = qualifier
    if runtime_user_id is not None:
        input_["runtime_user_id"] = runtime_user_id
    if trace_parent is not None:
        input_["trace_parent"] = trace_parent
    if trace_state is not None:
        input_["trace_state"] = trace_state
    if trace_id is not None:
        input_["trace_id"] = trace_id
    if baggage is not None:
        input_["baggage"] = baggage
    if model is not None:
        input_["model"] = model
    if system_prompt is not None:
        input_["system_prompt"] = system_prompt
    if tools is not None:
        input_["tools"] = tools
    if skills is not None:
        input_["skills"] = skills
    if allowed_tools is not None:
        input_["allowed_tools"] = allowed_tools
    if max_iterations is not None:
        input_["max_iterations"] = max_iterations
    if max_tokens is not None:
        input_["max_tokens"] = max_tokens
    if timeout_seconds is not None:
        input_["timeout_seconds"] = timeout_seconds
    if actor_id is not None:
        input_["actor_id"] = actor_id

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