cancel_document_checkout

Instructions

Implementation Reference

• src/cs_mcp_server/tools/documents.py:975-1055 (handler)

  The 'cancel_document_checkout' MCP tool handler function. It executes a GraphQL mutation 'cancelDocumentCheckout' to cancel a document checkout in the content repository using a reservation_id or document_id identifier.

  @mcp.tool(
      name="cancel_document_checkout",
  )
  async def cancel_document_checkout(
      identifier: str,
  ) -> Union[Document, ToolError]:
      """
      Cancels a document checkout in the content repository.
  
      :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).
  
      :returns: If successful, returns a Document object with its updated properties.
               If unsuccessful, returns a ToolError with details about the failure.
      """
      method_name = "cancel_document_checkout"
      try:
          # Prepare the mutation
          mutation = """
          mutation ($object_store_name: String!, $identifier: String!) {
            cancelDocumentCheckout(
              repositoryIdentifier: $object_store_name
              identifier: $identifier
            ) {
              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,
          }
  
          # Execute the GraphQL mutation
          logger.info("Executing document checkout cancellation")
          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 and return a Document instance from the response
          return Document.create_an_instance(
              graphQL_changed_object_dict=response["data"]["cancelDocumentCheckout"],
              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:975-977 (registration)

  Registration of the 'cancel_document_checkout' tool via the @mcp.tool decorator at line 976. The tool is registered within the 'register_document_tools' function.

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

• src/cs_mcp_server/mcp_server_main.py:32-270 (registration)

  The tool is registered in the main server via register_document_tools(mcp, graphql_client, metadata_cache) called for CORE and FULL server types.

  # Use absolute imports
  from cs_mcp_server.cache import MetadataCache
  from cs_mcp_server.client import GraphQLClient
  from cs_mcp_server.tools.documents import register_document_tools
  from cs_mcp_server.tools.classes import register_class_tools
  from cs_mcp_server.tools.search import (
      register_search_tools,
  )
  from cs_mcp_server.tools.legal_hold import register_hold_tools
  from cs_mcp_server.tools.vector_search import register_vector_search_tool
  from cs_mcp_server.tools.folders import register_folder_tools
  from cs_mcp_server.tools.annotations import register_annotation_tools
  from cs_mcp_server.tools.property_extraction import register_property_extraction_tools
  from cs_mcp_server.tools.classification import register_classification_tools
  from cs_mcp_server.resources.dynamic_resources import (
      register_dynamic_resources,
      DEFAULT_RESOURCE_FOLDERS,
  )
  from cs_mcp_server.tools.custom_objects import register_custom_object_tools
  from cs_mcp_server.tools.advanced_search import register_advanced_search_tools
  
  # Configure logging with dynamic level from environment variable
  log_level_name = os.environ.get("LOG_LEVEL", "INFO").upper()
  log_level = getattr(logging, log_level_name, logging.INFO)
  
  logging.basicConfig(
      level=log_level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  )
  logger = logging.getLogger(__name__)
  
  # Log the configured level for debugging
  logger.debug("Logging configured at %s level", log_level_name)
  
  # Global MCP instance - will be initialized by entry point
  mcp = None
  
  
  class ServerType(str, Enum):
      """Enumeration of available MCP server types."""
  
      CORE = "core"
      VECTOR_SEARCH = "vector-search"
      AI__DOCUMENT_INSIGHT = "ai-document-insight"
      LEGAL_HOLD = "legal-hold"
      PROPERTY_EXTRACTION_AND_CLASSIFICATION = "property-extraction-and-classification"
      FULL = "full"
  
  
  def _initialize_mcp_server(server_name: str) -> FastMCP:
      """
      Initialize the global MCP server instance.
  
      This function should be called once at the start of each entry point
      before any tool registration occurs.
  
      Args:
          server_name: The name for the MCP server instance
  
      Returns:
          FastMCP: The initialized MCP server instance
      """
      global mcp
      if mcp is None:
          mcp = FastMCP(server_name)
          logger.info("Initialized MCP server: %s", server_name)
      return mcp
  
  
  def parse_ssl_flag(value, default="true"):
      """
      Parse SSL flag which can be either a boolean or a path to a certificate.
  
      Args:
          value: The SSL flag value from environment variable
          default: Default value if not provided
  
      Returns:
          bool or str: True/False for boolean values, or the path string for certificates
      """
      if value is None:
          value = default
  
      # If it's a string representation of a boolean
      if value.lower() in ("true", "false"):
          return value.lower() == "true"
      # Otherwise it's a path to a certificate or other value
      return value
  
  
  def initialize_graphql_client():
      """
      Initialize the GraphQL client for the MCP server.
      Supports both basic authentication and OAuth authentication methods.
  
      Returns:
          GraphQLClient: The initialized GraphQL client instance
      """
      # Get connection details from environment variables
      graphql_url = os.environ.get("SERVER_URL", "")
      username = os.environ.get("USERNAME", "")
      password = os.environ.get("PASSWORD", "")
      ssl_enabled = parse_ssl_flag(os.environ.get("SSL_ENABLED"), "true")
      token_ssl_enabled = parse_ssl_flag(os.environ.get("TOKEN_SSL_ENABLED"), "true")
      object_store = os.environ.get("OBJECT_STORE", "")
      token_refresh = int(os.environ.get("TOKEN_REFRESH", "1800"))
  
      locale = os.environ.get("LOCALE", "")
      # OAuth specific parameters
      token_url = os.environ.get("TOKEN_URL", "")
      grant_type = os.environ.get("GRANT_TYPE", "")
      scope = os.environ.get("SCOPE", "")
      client_id = os.environ.get("CLIENT_ID", "")
      client_secret = os.environ.get("CLIENT_SECRET", "")
  
      # ZenIAM specific parameters
      zeniam_zen_url = os.environ.get("ZENIAM_ZEN_URL", "")
      zeniam_iam_url = os.environ.get("ZENIAM_IAM_URL", "")
      zeniam_iam_ssl_enabled = parse_ssl_flag(
          os.environ.get("ZENIAM_IAM_SSL_ENABLED"), "true"
      )
      zeniam_iam_grant_type = os.environ.get("ZENIAM_IAM_GRANT_TYPE", "")
      zeniam_iam_scope = os.environ.get("ZENIAM_IAM_SCOPE", "")
      zeniam_iam_client_id = os.environ.get("ZENIAM_IAM_CLIENT_ID", "")
      zeniam_iam_client_secret = os.environ.get("ZENIAM_IAM_CLIENT_SECRET", "")
      zeniam_iam_user_name = os.environ.get("ZENIAM_IAM_USER", "")
      zeniam_iam_user_password = os.environ.get("ZENIAM_IAM_PASSWORD", "")
      zeniam_zen_exchange_ssl = parse_ssl_flag(
          os.environ.get("ZENIAM_ZEN_SSL_ENABLED"), "true"
      )
  
      # Connection settings
      timeout = float(os.environ.get("REQUEST_TIMEOUT", "30.0"))
      pool_connections = int(os.environ.get("POOL_CONNECTIONS", "100"))
      pool_maxsize = int(os.environ.get("POOL_MAXSIZE", "100"))
  
      # Validate required parameters
      if not graphql_url:
          raise ValueError("SERVER_URL environment variable is required")
      if not username and not zeniam_zen_url:
          raise ValueError("USERNAME environment variable is required")
      if not password and not zeniam_zen_url:
          raise ValueError("PASSWORD environment variable is required")
      if not object_store:
          raise ValueError("OBJECT_STORE environment variable is required")
  
      # Create and return the GraphQL client
      # Pass all parameters to the constructor and let it handle them appropriately
      return GraphQLClient(
          url=graphql_url,
          username=username,
          password=password,
          ssl_enabled=ssl_enabled,
          object_store=object_store,
          token_url=token_url,
          token_ssl_enabled=token_ssl_enabled,
          grant_type=grant_type,
          scope=scope,
          client_id=client_id,
          client_secret=client_secret,
          timeout=timeout,
          locale=locale,
          pool_connections=pool_connections,
          pool_maxsize=pool_maxsize,
          token_refresh=token_refresh,
          ZenIAM_iam_url=zeniam_iam_url,
          ZenIAM_iam_ssl_enabled=zeniam_iam_ssl_enabled,
          ZenIAM_iam_grant_type=zeniam_iam_grant_type,
          ZenIAM_iam_scope=zeniam_iam_scope,
          ZenIAM_iam_client_id=zeniam_iam_client_id,
          ZenIAM_iam_client_secret=zeniam_iam_client_secret,
          ZenIAM_iam_user_name=zeniam_iam_user_name,
          ZenIAM_iam_user_password=zeniam_iam_user_password,
          ZenIAM_zen_url=zeniam_zen_url,
          ZenIAM_zen_exchange_ssl=zeniam_zen_exchange_ssl,
      )
  
  
  def register_server_resources(
      graphql_client: GraphQLClient,
      server_type: ServerType,
  ) -> None:
      """
      Register resources based on the server type.
  
      Resources provide read-only access to data and content for LLM context.
  
      Args:
          graphql_client: The initialized GraphQL client
          server_type: The type of server (ServerType enum)
      """
      # Ensure mcp is initialized (type narrowing for type checker)
      assert mcp is not None
  
      logger.info("Registering resources for %s server", server_type.value)
  
      # Register resources based on server type
      # Resources are now registered for CORE, AI_DOCUMENT_INSIGHT, LEGAL_HOLD,
      # PROPERTY_EXTRACTION_AND_CLASSIFICATION, and FULL server types
      if server_type in (
          ServerType.CORE,
          ServerType.AI__DOCUMENT_INSIGHT,
          ServerType.LEGAL_HOLD,
          ServerType.PROPERTY_EXTRACTION_AND_CLASSIFICATION,
          ServerType.FULL,
      ):
          # Get default folder for this server type
          default_folder = DEFAULT_RESOURCE_FOLDERS.get(server_type.value, "/resources")
  
          # Get dynamic resources folder path from environment, with server-specific default
          dynamic_resources_folder = os.environ.get("RESOURCES_FOLDER", default_folder)
  
          register_dynamic_resources(
              mcp, graphql_client, dynamic_resources_folder, server_type.value
          )
  
          logger.info("Resources registered from %s", dynamic_resources_folder)
  
  
  def register_server_tools(
      graphql_client: GraphQLClient,
      metadata_cache: MetadataCache,
      server_type: ServerType,
  ) -> None:
      """
      Register tools based on the server type.
  
      Args:
          graphql_client: The initialized GraphQL client
          metadata_cache: The metadata cache instance
          server_type: The type of server (ServerType enum)
      """
      # Ensure mcp is initialized (type narrowing for type checker)
      assert mcp is not None
  
      logger.info("Registering tools for %s server", server_type.value)
  
      # Register tools based on server type
      if server_type == ServerType.CORE:
          register_document_tools(mcp, graphql_client, metadata_cache)