checkout_document
Check out a document from the repository, optionally update its properties and download its content to a folder.
Instructions
Checks out a document in the content repository.
:param identifier: The document id or path (required). This can be either the document's ID (GUID) or its path in the repository (e.g., "/Folder1/document.pdf"). :param document_properties: Properties to update for the document during check-out. :param checkout_action: Check-out action parameters for the document. :param download_folder_path: Optional path to a folder where the document content will be downloaded. If not provided but content download is needed, the user will be prompted to provide it.
:returns: If successful, returns a Document object with its updated properties. If unsuccessful, returns a ToolError with details about the failure.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| identifier | Yes | ||
| document_properties | No | ||
| checkout_action | No | ||
| download_folder_path | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The handler function for the 'checkout_document' tool. It executes a GraphQL checkoutDocument mutation, processes optional document properties and checkout action, downloads content to a folder if requested, and returns a Document object or ToolError.
@mcp.tool( name="checkout_document", ) async def checkout_document( identifier: str, document_properties: Optional[DocumentPropertiesInput] = None, checkout_action: Optional[SubCheckoutActionInput] = None, download_folder_path: Optional[str] = None, ) -> Union[Document, ToolError]: """ Checks out a document in the content repository. :param identifier: The document id or path (required). This can be either the document's ID (GUID) or its path in the repository (e.g., "/Folder1/document.pdf"). :param document_properties: Properties to update for the document during check-out. :param checkout_action: Check-out action parameters for the document. :param download_folder_path: Optional path to a folder where the document content will be downloaded. If not provided but content download is needed, the user will be prompted to provide it. :returns: If successful, returns a Document object with its updated properties. If unsuccessful, returns a ToolError with details about the failure. """ method_name = "checkout_document" try: # Prepare the mutation mutation = """ mutation ($object_store_name: String!, $identifier: String!, $document_properties: DocumentPropertiesInput, $checkout_action: SubCheckoutActionInput) { checkoutDocument( repositoryIdentifier: $object_store_name identifier: $identifier documentProperties: $document_properties checkoutAction: $checkout_action ) { id className reservation{ isReserved id } currentVersion{ contentElements{ ... on ContentTransferType { retrievalName contentType contentSize downloadUrl } } } properties { id value } } } """ # Prepare variables for the GraphQL query variables = { "object_store_name": graphql_client.object_store, "identifier": identifier, "document_properties": None, "checkout_action": None, } # Process document properties if provided if document_properties: try: document_properties.eval() transformed_props = document_properties.transform_properties_dict( exclude_none=True ) variables["document_properties"] = transformed_props except Exception as e: logger.error("Error transforming document properties: %s", str(e)) logger.error(traceback.format_exc()) return ToolError( message=f"{method_name} failed: {str(e)}. Trace available in server logs." ) # Handle checkout action if provided if checkout_action: # Use model_dump with exclude_none for cleaner code variables["checkout_action"] = checkout_action.model_dump( exclude_none=True ) # Execute the GraphQL mutation logger.info("Executing document check-out") response: Union[ToolError, Dict[str, Any]] = ( await graphql_client_execute_async_wrapper( logger, method_name, graphql_client, query=mutation, variables=variables, ) ) if isinstance(response, ToolError): return response # Create a Document instance from the response document = Document.create_an_instance( graphQL_changed_object_dict=response["data"]["checkoutDocument"], class_identifier=DEFAULT_DOCUMENT_CLASS, ) # Check if we need to download content if ( download_folder_path and "currentVersion" in response["data"]["checkoutDocument"] ): content_elements = response["data"]["checkoutDocument"][ "currentVersion" ]["contentElements"] if content_elements and len(content_elements) > 0: logger.info( "Found %s content elements to download", len(content_elements) ) download_results = [] download_errors = [] for idx, element in enumerate(content_elements): if "downloadUrl" in element and element["downloadUrl"]: download_url = element["downloadUrl"] logger.info( "Downloading content element %s/%s: %s", idx + 1, len(content_elements), element["retrievalName"], ) download_result = ( await graphql_client.download_content_async( download_url=download_url, download_folder_path=download_folder_path, ) ) if download_result["success"]: download_results.append(download_result) logger.info( "Content element %s downloaded to %s", idx + 1, download_result["file_path"], ) else: error_msg = ( "Failed to download content element %s: %s" % ( idx + 1, download_result["error"], ) ) download_errors.append(error_msg) logger.warning(error_msg) if download_errors: error_message = ( "Document checkout was successful, but %s content downloads failed: %s" % (len(download_errors), "; ".join(download_errors)) ) logger.warning(error_message) return ToolError( message=error_message, suggestions=[ "Check if the download folder exists and is writable", "Verify network connectivity to the content server", "Try downloading the files without checking out the document", ], ) elif download_results: logger.info( "Successfully downloaded %s content elements", len(download_results), ) return document except Exception as e: logger.error("%s failed: %s", method_name, str(e)) logger.error(traceback.format_exc()) return ToolError( message=f"{method_name} failed: {str(e)}. Trace available in server logs." ) - The SubCheckoutActionInput schema used by checkout_document for defining checkout action parameters (reservationId, reservationType, reservationClass, reservationProperties, reservationObjectProperties).
class SubCheckoutActionInput(BaseModel): """Input for document check-out action.""" reservationId: Optional[str] = Field( default=None, description="ID for the reservation" ) reservationType: Optional[ReservationType] = Field( default=None, description="Type of reservation (COLLABORATIVE, EXCLUSIVE, or OBJECT_STORE_DEFAULT)", ) reservationClass: Optional[str] = Field( default=None, description="Class for the reservation" ) reservationProperties: Optional[List[PropertyIdentifierAndScalarValue]] = Field( default=None, description="Properties for the reservation" ) reservationObjectProperties: Optional[List[ObjectPropertyInput]] = Field( default=None, description="Object properties for the reservation" ) - The DocumentPropertiesInput schema used by checkout_document for defining document properties during checkout.
class DocumentPropertiesInput(CustomInputBase): """Input for document properties.""" properties: Optional[List[PropertyIdentifierAndScalarValue]] = Field( default=None, description="Properties for Document" ) name: Optional[str] = Field( default=None, description="Name sets DocumentTitle or whatever property is configured as the Name property", ) owner: Optional[str] = Field(default=None, description="Owner") content: Optional[str] = Field( default=None, description="Content can be specified if this represents a Reservation document or document creation", ) mimeType: Optional[str] = Field(default=None, description="Mime type") compoundDocumentState: Optional[str] = Field( default=None, description="Compound document state" ) cmRetentionDate: Optional[datetime] = Field( default=None, description="Retention date" ) # contentElements field removed from the model to prevent agents from interpreting and creating this field # Instead, we use the methods from CustomInputBase to add content elements programmatically # Commented out references to ObjectReferenceInput, PermissionListInput, ObjectPropertyInput """ objectProperties: Optional[List[ObjectPropertyInput]] = Field( default=None, description="Object properties" ) replicationGroup: Optional[ObjectReferenceInput] = Field( default=None, description="Replication group" ) permissions: Optional[PermissionListInput] = Field( default=None, description="Permissions" ) securityPolicy: Optional[ObjectReferenceInput] = Field( default=None, description="Security policy" ) securityFolder: Optional[ObjectReferenceInput] = Field( default=None, description="Security folder" ) storagePolicy: Optional[ObjectReferenceInput] = Field( default=None, description="Storage policy" ) documentLifecyclePolicy: Optional[ObjectReferenceInput] = Field( default=None, description="Document lifecycle policy" ) storageArea: Optional[ObjectReferenceInput] = Field( default=None, description="Storage area" ) """ - src/cs_mcp_server/tools/documents.py:60-62 (registration)The register_document_tools function that registers the checkout_document tool (and other document tools) via the @mcp.tool decorator.
def register_document_tools( mcp: FastMCP, graphql_client: GraphQLClient, metadata_cache: MetadataCache ) -> None: - The graphql_client_execute_async_wrapper helper function used by checkout_document for executing GraphQL queries with error handling, timing, and logging.
async def graphql_client_execute_async_wrapper ( logger: Logger, method_name: str, graphql_client: GraphQLClient, query: str, variables: Optional[Dict[str, Any]] = None ) -> Union [ToolError, Dict[str, Any]]: "Wrapper for graphql_client.execute_async to handle errors, timing and logging of GraphQL queries." start_time = time.perf_counter() response = None try: logger.debug(f"{method_name}, GraphQL query: {query}, GraphQL variables: {variables} ") response = await graphql_client.execute_async(query=query, variables=variables) if "errors" in response: error_message = response["errors"] logger.error(f"{method_name} failed: {error_message}") return ToolError( message=f"{method_name} failed: got err {error_message}. Trace available in server logs.", ) if "error" in response: error_type = response.get("error_type", "") # Get error_type if it exists, otherwise empty string error_message = f"error_type = {error_type}, message = {response["message"]}" logger.error(f"{method_name} failed: {error_message}") return ToolError( message=f"{method_name} failed: got err {error_message}. Trace available in server logs.", ) if "data" not in response or response["data"] is None: error_message = f" No 'data' returned from GraphQL query" logger.error(f"{method_name} failed: {error_message}") return ToolError( message=f"{method_name} failed: got err {error_message}. Trace available in server logs.", ) return response except Exception as ex: error_traceback = traceback.format_exc(limit=TRACEBACK_LIMIT) logger.error( f"{method_name} failed: {ex.__class__.__name__} - {str(ex)}\n{error_traceback}" ) return ToolError( message=f"{method_name} failed: got err {ex}. Trace available in server logs.", ) finally: logger.debug(f"{method_name}, GraphQL response (elapse {time.perf_counter() - start_time:.2f}s): {response}")