create_policy_engine ¶

def create_policy_engine(
    self,
    name: "capo_bedrock_agentcore_control.types.policy_engine_name.PolicyEngineName",
    *,
    config_overrides: Optional[BedrockAgentCoreControlClientConfig] = None,
    description: Optional[
        "capo_bedrock_agentcore_control.types.description.Description"
    ] = None,
    client_token: Optional[
        "capo_bedrock_agentcore_control.types.client_token.ClientToken"
    ] = None,
    encryption_key_arn: Optional[
        "capo_bedrock_agentcore_control.types.kms_key_arn.KmsKeyArn"
    ] = None,
    tags: Optional["capo_bedrock_agentcore_control.types.tags_map.TagsMap"] = None,
) -> "capo_bedrock_agentcore_control.types.create_policy_engine_response.CreatePolicyEngineResponse":
    """<p>Creates a new policy engine within the AgentCore Policy system. A policy engine is a collection of policies that evaluates and authorizes agent tool calls. When associated with Gateways (each Gateway can be associated with at most one policy engine, but multiple Gateways can be associated with the same engine), the policy engine intercepts all agent requests and determines whether to allow or deny each action based on the defined policies. This is an asynchronous operation. Use the <a href="https://docs.aws.amazon.com/bedrock-agentcore-control/latest/APIReference/API_GetPolicyEngine.html">GetPolicyEngine</a> operation to poll the <code>status</code> field to track completion.</p>

    Args:
        name: <p>The customer-assigned immutable name for the policy engine. This name identifies the policy engine and cannot be changed after creation.</p>
        description: <p>A human-readable description of the policy engine's purpose and scope (1-4,096 characters). This helps administrators understand the policy engine's role in the overall governance strategy. Document which Gateway this engine will be associated with, what types of tools or workflows it governs, and the team or service responsible for maintaining it. Clear descriptions are essential when managing multiple policy engines across different services or environments.</p>
        client_token: <p>A unique, case-sensitive identifier that you provide to ensure the idempotency of the request. If you retry a request with the same client token, the service returns the same response without creating a duplicate policy engine.</p>
        encryption_key_arn: <p>The Amazon Resource Name (ARN) of the KMS key used to encrypt the policy engine data.</p>
        tags: <p>A map of tag keys and values to assign to an AgentCore Policy. Tags enable you to categorize your resources in different ways, for example, by purpose, owner, or environment.</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.
    """

    def _handler(
        req: "OperationRequest[capo_bedrock_agentcore_control.types.create_policy_engine_request.CreatePolicyEngineRequest]",
    ) -> OperationResponse[
        "capo_bedrock_agentcore_control.types.create_policy_engine_response.CreatePolicyEngineResponse"
    ]:
        import capo_bedrock_agentcore_control._operations.amazon_bedrock_agent_core_control.create_policy_engine

        output, http_response = (
            capo_bedrock_agentcore_control._operations.amazon_bedrock_agent_core_control.create_policy_engine.create_policy_engine(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

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

        return capo_bedrock_agentcore_control.types.create_policy_engine_request.redact_sensitive(
            input_
        )

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

        return capo_bedrock_agentcore_control.types.create_policy_engine_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "CreatePolicyEngine",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_agentcore_control.types.create_policy_engine_request.CreatePolicyEngineRequest = {
        "name": name
    }
    if description is not None:
        input_["description"] = description
    if client_token is None:
        client_token = str(uuid.uuid4())
    input_["client_token"] = client_token
    if encryption_key_arn is not None:
        input_["encryption_key_arn"] = encryption_key_arn
    if tags is not None:
        input_["tags"] = tags

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    response.response.close()
    return response.output