create_registry_record ¶

def create_registry_record(
    self,
    registry_id: "capo_agent_registry_control.types.registry_identifier.RegistryIdentifier",
    name: "capo_agent_registry_control.types.registry_record_name.RegistryRecordName",
    record_type: "capo_agent_registry_control.types.record_type.RecordType",
    descriptors: "capo_agent_registry_control.types.descriptors.Descriptors",
    *,
    config_overrides: Optional[AgentRegistryControlClientConfig] = None,
    display_name: Optional[
        "capo_agent_registry_control.types.registry_record_display_name.RegistryRecordDisplayName"
    ] = None,
    description: Optional[
        "capo_agent_registry_control.types.description.Description"
    ] = None,
    record_version: Optional[
        "capo_agent_registry_control.types.registry_record_version.RegistryRecordVersion"
    ] = None,
    client_token: Optional[
        "capo_agent_registry_control.types.client_token.ClientToken"
    ] = None,
    provenance: Optional[
        "capo_agent_registry_control.types.provenance_list.ProvenanceList"
    ] = None,
    custom_metadata: Optional[
        "capo_agent_registry_control.types.custom_metadata_document.CustomMetadataDocument"
    ] = None,
    tags: Optional["capo_agent_registry_control.types.tags_map.TagsMap"] = None,
) -> "capo_agent_registry_control.types.create_registry_record_response.CreateRegistryRecordResponse":
    """<p>Creates a registry record within a registry. A registry record describes a discoverable resource, such as an MCP server, an agent, an agent skill, or a custom resource. Creation is asynchronous: the record is returned with the CREATING status while it is processed.</p>

    Args:
        registry_id: <p>The identifier of the registry in which to create the record (ARN or ID)</p>
        name: <p>The name of the registry record</p>
        display_name: <p>The human-readable display name of the registry record</p>
        description: <p>The description of the registry record</p>
        record_type: <p>The type of the registry record, which determines the descriptor format</p>
        descriptors: <p>The typed descriptor content for the registry record</p>
        record_version: <p>The version of the registry record</p>
        client_token: <p>Client token for idempotency</p>
        provenance: <p>The provenance lineage entries for the registry record. This field is reserved for the Amazon Web Services Agent Registry auto-detection service principal. Requests that include this field from other callers are rejected.</p>
        custom_metadata: <p>The custom metadata for the record. Values are strings or JSON booleans.</p>
        tags: <p>Tags to associate with the registry record</p>

    Raises:
        capo_agent_registry_control.errors.access_denied_exception.AccessDeniedException: <p>The caller is not authorized to perform the requested action.</p>
        capo_agent_registry_control.errors.conflict_exception.ConflictException: <p>The request conflicts with the current state of the resource.</p>
        capo_agent_registry_control.errors.internal_server_exception.InternalServerException: <p>The request failed due to an unexpected internal error; the caller may retry.</p>
        capo_agent_registry_control.errors.resource_not_found_exception.ResourceNotFoundException: <p>The requested resource was not found.</p>
        capo_agent_registry_control.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: <p>The request would exceed a service quota.</p>
        capo_agent_registry_control.errors.throttling_exception.ThrottlingException: <p>The request was denied due to request throttling; the caller may retry after a delay.</p>
        capo_agent_registry_control.errors.validation_exception.ValidationException: <p>The request failed validation of one or more input fields.</p>
        capo_agent_registry_control.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_agent_registry_control.types.create_registry_record_request.CreateRegistryRecordRequest]",
    ) -> OperationResponse[
        "capo_agent_registry_control.types.create_registry_record_response.CreateRegistryRecordResponse"
    ]:
        import capo_agent_registry_control._operations.agent_registry_control.create_registry_record

        output, http_response = (
            capo_agent_registry_control._operations.agent_registry_control.create_registry_record.create_registry_record(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_agent_registry_control.types.create_registry_record_request.CreateRegistryRecordRequest",
    ) -> dict[str, Any]:
        import capo_agent_registry_control.types.create_registry_record_request

        return capo_agent_registry_control.types.create_registry_record_request.redact_sensitive(
            input_
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("CreateRegistryRecord", redact_input=_redact_input),
    )
    input_: capo_agent_registry_control.types.create_registry_record_request.CreateRegistryRecordRequest = {
        "registry_id": registry_id,
        "name": name,
        "record_type": record_type,
        "descriptors": descriptors,
    }
    if display_name is not None:
        input_["display_name"] = display_name
    if description is not None:
        input_["description"] = description
    if record_version is not None:
        input_["record_version"] = record_version
    if client_token is None:
        client_token = str(uuid.uuid4())
    input_["client_token"] = client_token
    if provenance is not None:
        input_["provenance"] = provenance
    if custom_metadata is not None:
        input_["custom_metadata"] = custom_metadata
    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