IBM Core Content Services MCP Server

Instructions

Implementation Reference

• src/cs_mcp_server/tools/documents.py:579-714 (handler)

  The core handler function for the 'checkin_document' tool. It performs a GraphQL 'checkinDocument' mutation to check in the specified document or reservation, optionally updating properties and uploading files. Handles errors and returns a Document object or ToolError.

  async def checkin_document(
      identifier: str,
      checkin_action: Optional[SubCheckinActionInput] = SubCheckinActionInput(),
      document_properties: Optional[DocumentPropertiesInput] = None,
      file_paths: Optional[List[str]] = None,
  ) -> Union[Document, ToolError]:
      """
      Checks in a document in the content repository with specified properties.
  
      :param identifier: The identifier (required). This can be either a reservation_id or document_id.
                        Reservation ID (GUID) is prioritized.
                        Otherwise, we use document_id (GUID).
      :param checkin_action: Check-in action parameters for the document.
      :param document_properties: Properties to update for the document during check-in.
      :param file_paths: Optional list of file paths to upload as the document's content.
  
      :returns: If successful, returns a Document object with its updated properties.
               If unsuccessful, returns a ToolError with details about the failure.
      """
      method_name = "checkin_document"
      try:
          # Prepare the mutation
          mutation = """
          mutation ($object_store_name: String!, $identifier: String!,
                   $document_properties: DocumentPropertiesInput, $checkin_action: SubCheckinActionInput!) {
            checkinDocument(
              repositoryIdentifier: $object_store_name
              identifier: $identifier
              documentProperties: $document_properties
              checkinAction: $checkin_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,
              "checkin_action": None,
          }
  
          # Process file paths
          file_paths_dict = {}
  
          # Handle file upload if file paths are provided
          if file_paths:
              try:
                  # Initialize document_properties if not provided
                  if not document_properties:
                      document_properties = DocumentPropertiesInput()
  
                  file_paths_dict = document_properties.process_file_content(
                      file_paths
                  )
              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."
                  )
  
          # Process document properties if provided
          if document_properties:
              try:
                  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."
                  )
  
          if checkin_action:
              # Handle checkin action if provided                # Use model_dump with exclude_none for cleaner code
              variables["checkin_action"] = checkin_action.model_dump(
                  exclude_none=True
              )
  
          # Execute the GraphQL mutation
          if file_paths_dict:
              # Use execute with file_paths for file upload
              logger.info("Executing document check-in with file upload")
              response = graphql_client.execute(
                  query=mutation,
                  variables=variables,
                  file_paths=file_paths_dict,
              )
          else:
              # Use execute_async for regular document check-in
              logger.info("Executing document check-in")
              response = await graphql_client.execute_async(
                  query=mutation, variables=variables
              )
  
          # Handle errors
          if "errors" in response:
              logger.error("GraphQL error: %s", response["errors"])
              return ToolError(message=f"{method_name} failed: {response['errors']}")
  
          # Create and return a Document instance from the response
          return Document.create_an_instance(
              graphQL_changed_object_dict=response["data"]["checkinDocument"],
              class_identifier=DEFAULT_DOCUMENT_CLASS,
          )
  
      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."
          )

• src/cs_mcp_server/tools/documents.py:576-578 (registration)

  The @mcp.tool decorator registers the 'checkin_document' handler function within the register_document_tools function.

  @mcp.tool(
      name="checkin_document",
  )

• src/cs_mcp_server/mcp_server_main.py:219-225 (registration)

  Calls register_document_tools to register document tools (including checkin_document) for CORE server type in register_server_tools function.

  if server_type == ServerType.CORE:
      register_document_tools(mcp, graphql_client, metadata_cache)
      register_folder_tools(mcp, graphql_client)
      register_class_tools(mcp, graphql_client, metadata_cache)
      register_search_tools(mcp, graphql_client, metadata_cache)
      register_annotation_tools(mcp, graphql_client)
      logger.info("Core tools registered")

• src/cs_mcp_server/mcp_server_main.py:235-243 (registration)

  Calls register_document_tools to register document tools (including checkin_document) for FULL server type in register_server_tools function.

  elif server_type == ServerType.FULL:
      register_document_tools(mcp, graphql_client, metadata_cache)
      register_folder_tools(mcp, graphql_client)
      register_class_tools(mcp, graphql_client, metadata_cache)
      register_search_tools(mcp, graphql_client, metadata_cache)
      register_annotation_tools(mcp, graphql_client)
      register_vector_search_tool(mcp, graphql_client)
      register_legalhold(mcp, graphql_client)
      logger.info("All tools registered")

• src/cs_mcp_server/tools/documents.py:579-583 (schema)

  Type hints define the input schema: identifier (str), checkin_action (SubCheckinActionInput), document_properties (DocumentPropertiesInput), file_paths (list[str]). Output: Document or ToolError.

  async def checkin_document(
      identifier: str,
      checkin_action: Optional[SubCheckinActionInput] = SubCheckinActionInput(),
      document_properties: Optional[DocumentPropertiesInput] = None,
      file_paths: Optional[List[str]] = None,