create_harness async ¶

async def create_harness(
    self,
    harness_name: "capo_bedrock_agentcore_control.types.harness_name.HarnessName",
    execution_role_arn: "capo_bedrock_agentcore_control.types.role_arn.RoleArn",
    *,
    config_overrides: Optional[AsyncBedrockAgentCoreControlClientConfig] = None,
    client_token: Optional[
        "capo_bedrock_agentcore_control.types.client_token.ClientToken"
    ] = None,
    environment: Optional[
        "capo_bedrock_agentcore_control.types.harness_environment_provider_request.HarnessEnvironmentProviderRequest"
    ] = None,
    environment_artifact: Optional[
        "capo_bedrock_agentcore_control.types.harness_environment_artifact.HarnessEnvironmentArtifact"
    ] = None,
    environment_variables: Optional[
        "capo_bedrock_agentcore_control.types.environment_variables_map.EnvironmentVariablesMap"
    ] = None,
    authorizer_configuration: Optional[
        "capo_bedrock_agentcore_control.types.authorizer_configuration.AuthorizerConfiguration"
    ] = None,
    model: Optional[
        "capo_bedrock_agentcore_control.types.harness_model_configuration.HarnessModelConfiguration"
    ] = None,
    system_prompt: Optional[
        "capo_bedrock_agentcore_control.types.harness_system_prompt.HarnessSystemPrompt"
    ] = None,
    tools: Optional[
        "capo_bedrock_agentcore_control.types.harness_tools.HarnessTools"
    ] = None,
    skills: Optional[
        "capo_bedrock_agentcore_control.types.harness_skills.HarnessSkills"
    ] = None,
    allowed_tools: Optional[
        "capo_bedrock_agentcore_control.types.harness_allowed_tools.HarnessAllowedTools"
    ] = None,
    memory: Optional[
        "capo_bedrock_agentcore_control.types.harness_memory_configuration.HarnessMemoryConfiguration"
    ] = None,
    truncation: Optional[
        "capo_bedrock_agentcore_control.types.harness_truncation_configuration.HarnessTruncationConfiguration"
    ] = None,
    hooks: Optional[
        "capo_bedrock_agentcore_control.types.harness_hooks.HarnessHooks"
    ] = None,
    max_iterations: Optional[int] = None,
    max_tokens: Optional[int] = None,
    timeout_seconds: Optional[int] = None,
    tags: Optional["capo_bedrock_agentcore_control.types.tags_map.TagsMap"] = None,
) -> "capo_bedrock_agentcore_control.types.create_harness_response.CreateHarnessResponse":
    """<p>Operation to create a harness.</p>

    Args:
        harness_name: <p>The name of the harness. Must start with a letter and contain only alphanumeric characters and underscores.</p>
        client_token: <p>A unique, case-sensitive identifier to ensure idempotency of the request.</p>
        execution_role_arn: <p>The ARN of the IAM role that the harness assumes when running. This role must have permissions for the services the agent needs to access, such as Amazon Bedrock for model invocation.</p>
        environment: <p>The compute environment configuration for the harness, including network and lifecycle settings.</p>
        environment_artifact: <p>The environment artifact for the harness, such as a custom container image containing additional dependencies.</p>
        environment_variables: <p>Environment variables to set in the harness runtime environment.</p>
        model: <p>The model configuration for the harness. Supports Amazon Bedrock, OpenAI, and Google Gemini model providers.</p>
        system_prompt: <p>The system prompt that defines the agent's behavior and instructions.</p>
        tools: <p>The tools available to the agent, such as remote MCP servers, AgentCore Gateway, AgentCore Browser, Code Interpreter, or inline functions.</p>
        skills: <p>The skills available to the agent. Skills are bundles of files that the agent can pull into its context on demand.</p>
        allowed_tools: <p>The tools that the agent is allowed to use. Supports glob patterns such as * for all tools, @builtin for all built-in tools, or @serverName/toolName for specific MCP server tools.</p>
        memory: <p>The AgentCore Memory configuration for persisting conversation context across sessions.</p>
        truncation: <p>The truncation configuration for managing conversation context when it exceeds model limits.</p>
        hooks: <p>The lifecycle hooks to run at defined points in the agent loop.</p>
        max_iterations: <p>The maximum number of iterations the agent loop can execute per invocation.</p>
        max_tokens: <p>The maximum total number of output tokens the agent can generate across all model calls within a single invocation.</p>
        timeout_seconds: <p>The maximum duration in seconds for the agent loop execution per invocation.</p>
        tags: <p>Tags to apply to the harness resource.</p>

    Raises:
        capo_bedrock_agentcore_control.errors.access_denied_exception.AccessDeniedException: <p>This exception is thrown when a request is denied per access permissions</p>
        capo_bedrock_agentcore_control.errors.conflict_exception.ConflictException: <p>This exception is thrown when there is a conflict performing an operation</p>
        capo_bedrock_agentcore_control.errors.internal_server_exception.InternalServerException: <p>This exception is thrown if there was an unexpected error during processing of request</p>
        capo_bedrock_agentcore_control.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: <p>This exception is thrown when a request is made beyond the service quota</p>
        capo_bedrock_agentcore_control.errors.throttling_exception.ThrottlingException: <p>This exception is thrown when the number of requests exceeds the limit</p>
        capo_bedrock_agentcore_control.errors.validation_exception.ValidationException: <p>The input fails to satisfy the constraints specified by the service.</p>
        capo_bedrock_agentcore_control.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_bedrock_agentcore_control.types.create_harness_request.CreateHarnessRequest]",
    ) -> AsyncOperationResponse[
        "capo_bedrock_agentcore_control.types.create_harness_response.CreateHarnessResponse"
    ]:
        import capo_bedrock_agentcore_control._operations.amazon_bedrock_agent_core_control.create_harness

        (
            output,
            http_response,
        ) = await capo_bedrock_agentcore_control._operations.amazon_bedrock_agent_core_control.create_harness.async_create_harness(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_bedrock_agentcore_control.types.create_harness_request.CreateHarnessRequest",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore_control.types.create_harness_request

        return capo_bedrock_agentcore_control.types.create_harness_request.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_bedrock_agentcore_control.types.create_harness_response.CreateHarnessResponse",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore_control.types.create_harness_response

        return capo_bedrock_agentcore_control.types.create_harness_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "CreateHarness",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_agentcore_control.types.create_harness_request.CreateHarnessRequest = {
        "harness_name": harness_name,
        "execution_role_arn": execution_role_arn,
    }
    if client_token is None:
        client_token = str(uuid.uuid4())
    input_["client_token"] = client_token
    if environment is not None:
        input_["environment"] = environment
    if environment_artifact is not None:
        input_["environment_artifact"] = environment_artifact
    if environment_variables is not None:
        input_["environment_variables"] = environment_variables
    if authorizer_configuration is not None:
        input_["authorizer_configuration"] = authorizer_configuration
    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 memory is not None:
        input_["memory"] = memory
    if truncation is not None:
        input_["truncation"] = truncation
    if hooks is not None:
        input_["hooks"] = hooks
    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 tags is not None:
        input_["tags"] = tags

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output