Sitemap
Open in app

Sign in

Medium Logo
Write
Search

Sign in

Musings of a software architect

Musings of a software architect

I‘ve worked for IBM all of my career and I am an avid technologist who is keen to get his hands dirty. My role affords me this opportunity and I share what I can

Batch Processing FileNet Documents with MCP Tools

Tony Hickman
18 min readJan 9, 2026

--

Often “Batch” is seen as a bad term but there are still a LOT of areas where “batch” processing is the sensible architectural approach. Recently I was asked to explore how a recently GA’d set of FileNet MCP Tools from the IBM FileNet development team (available at https://github.com/ibm-ecm/ibm-content-services-mcp-server) could be used to process a “batch” of documents stored in FileNet. In this post I’ll cover the architecture I used and go into some details around the problem statement and the technical approach.

Problem Statement

Often large enterprise content management deployments (like FileNet) contain a lot of information which is poorly classified and/or does not have valuable meta-data extracted to support searching. Given this I was asked to develop an approach which would handle the processing of a set of identified documents from FileNet. These documents needed to be processed as follows:

For Each Document in Batch:

  1. Get Documents extract text
  2. Extract the top 3 keywords in the text
  3. Use the keywords to determine the document class to assign the document to
  4. Update the document to set it to the determined document class
  5. Get class-specific properties for the assigned document class
  6. Try and extract values for these properties from the text
  7. Update document properties with any extracted values

Architeture Overview

A key architectural design point was to leverage capabilities from FileNet and the MCP Tools and not re-implement these. Based on this I defined the following architecture.

Architecture Overview

Let me break this down a bit…

Firstly, the approach to “trigger” the batch would be via “sweeps” within FileNet coupled with a new “sweep action”. The “sweep” is responsible for identifying the set of documents to be processed. This is standard functionality and more details can be found here. The “sweep” would use the “sweep action” to call a REST endpoint in the Controller passing the collection of documents that make up the batch.

Next, the Controller would store the details of the batch in a database and get an identifier for this record in the database.

The Controller would then start a “job” in OpenShift [passing the database record identifier] to process the batch.

The OpenShift “job” would the use a combination of the FileNet MCP Tools (show as the connection between the Batch Job and FileNet) and LLM prompts (hosted in watsonx.ai) to perform the necessary processing steps.

With that covered I will now go into more detail on each component.

Technical Details

For anyone who subscribes to my posts, this approach will be familiar as its basically the same as the one used for my work on generating rules in IBM ADS (see here). Re-use always good :-)

Batch Controller

The Batch Controller uses `FastAPI` to present a REST POST endpoint \run to trigger a batch run. The POST provides a JSON payload of the following format to specify the `batch` to run.

{
   "documents": [
     {
       "identifier": "string"
     }
   ]
}

This endpoint is secured via a secret token which is provided in the `X-CILENT-SECRET` in the HTTP Request header. Once a `batch` has been received a `Job` is created to run the generation process for the specified triage and the `jobID` for the job is returned.

Three further REST endpoints are provided to retrieve information about the build jobs (these are all secured with the `X-CILENT-SECRET`):

  • GET /jobs : returns details of the jobs that are maintained in OpenShift project (NB: It is possible to remove completed and failed jobs via the OpenShift Console or CLI)
  • GET /job/{jobID} : returns details of the job with the specified `jobID`
  • GET /job/{jobID}/status : returns status details of the job with the specified `jobID`

FileNet Batch (Job)

The Python control program is contained with `run_batch.py`. This program is run via an OpenShift `Job` which is created by the FastAPI app running in the Controller container.

The program requires the following environment variables to configure its self:

  • LLM_API_KEY the IAM key to allow prompts to be executed in watsonx.ai
  • ASTRABD_TOKEN token to allow access to AstrDB
  • ASTRA_ENDPOINT AstraDb endpoint
  • CPBA_USERNAME user to use to access FileNet environment
  • CPBA_PASSWORD password to use to access FileNet environment
  • CPBA_SERVER_URL FileNet graphQL endpoint
  • CPBA_CONTENT_STORE FileNet content store which contains the documents to be processed

When the job has started the id passed as a shell parameter is used to retrieve the batch definition from AstraDb. This is then passed to the run() function to process the batch. The run() function first instantiates LLMHandler and MCPHandlerobjects. It then iterates over the collection of documents and calls process_doc() to perform the necessary processing.

Get Tony Hickman’s stories in your inbox

Join Medium for free to get updates from this writer.

The process_doc() function performs the processing steps documented above.

Batch Data Store

To address the technical requirements for the Data Store I selected AstraDb. This was primarily due to the fact I had code to hand to allow me to rapidly implement the necessary functionality. Fundamentally any style of database could be used here [you could even use Cloud Object Storage], though a NoSQL style of database does lend itself well to this use case. The key thing is to be able to get a unique identifier to the stored batch details that can be passed to the Batch Job.

Sweep Action

I am very much a novice on FileNet sweep actions, so I used IBM Bob to get me started. I decided to go for a JavaScript action rather than Java as it was easier for me to create. After a bit of cleaning up I arrived at the following.

importPackage(Packages.com.filenet.api.core);
importPackage(Packages.com.filenet.api.constants);
importPackage(Packages.com.filenet.api.exception);
importPackage(Packages.com.filenet.api.sweep);
importPackage(Packages.com.filenet.api.engine);
importPackage(Packages.java.net);
importPackage(Packages.java.io);
importPackage(Packages.java.lang);

function getRequiredProperties(){
 return ""
}

function onPolicySweep(SweepObject, SweepPolicy, SweepItems){
}

function onSweep(SweepObject, SweepItems){
  var hcc = HandlerCallContext.getInstance();
  hcc.traceDetail("Entering EmptyHandler.onSweep");

var ids = [];
  for (var i = 0; i < SweepItems.length; i++) {
    // Abort cleanly if server is shutting down
    if (hcc != null && hcc.isShuttingDown()) {
      throw new EngineRuntimeException(
        ExceptionCode.E_BACKGROUND_TASK_TERMINATED,
        "Terminating: server is shutting down");
    }

    try {
      var target = SweepItems[i].getTarget(); // IndependentObject (Document)
      var idVal = target.getProperties().getIdValue("Id").toString();
      var rec = {"identifier" : idVal}
      ids.push(rec);

      // Mark item processed (our action is "reporting", not mutating)
      SweepItems[i].setOutcome(SweepItemOutcome.PROCESSED,
                               "Id collected: " + idVal);
    } catch (e) {
      // Mark the specific item as failed, but continue
      SweepItems[i].setOutcome(SweepItemOutcome.FAILED,
                               "Failed to collect Id: " + e);
    }
  }

  body = {
    "documents": ids
  };


  httpPostJson("https://<OCP Route>/run", body, "x-client-secret","<secret>", hcc)
}

// ----------------- HTTP helper -----------------
function httpPostJson(urlStr, jsonBody, authHeaderName, authHeaderVal, hcc) {
  var url = new URL(urlStr);
  var conn = url.openConnection();
  conn.setDoOutput(true);
  conn.setRequestMethod("POST");
  conn.setRequestProperty("Content-Type", "application/json");
  if (authHeaderName && authHeaderVal) {
    conn.setRequestProperty(authHeaderName, authHeaderVal);
  }

  var out = new OutputStreamWriter(conn.getOutputStream(), "UTF-8");
  out.write(JSON.stringify(jsonBody));
  out.flush();
  out.close();

  var responseCode = conn.getResponseCode();
  if (responseCode >= 200 && responseCode < 300) {
    var reader = new BufferedReader(new InputStreamReader(conn.getInputStream(), "UTF-8"));
    var line, resp = "";
    while ((line = reader.readLine()) != null) { resp += line; }
    reader.close();
    hcc.traceDetail("POST OK (" + responseCode + "): " + resp);
  } else {
    var err = conn.getErrorStream();
    var msg = "POST failed (" + responseCode + ")";
    if (err != null) {
      var r = new BufferedReader(new InputStreamReader(err, "UTF-8"));
      var ln, buf = "";
      while ((ln = r.readLine()) != null) { buf += ln; }
      r.close();
      msg += " body=" + buf;
    }
    // Log and let the sweep continue; you can switch this to throw if you want retries
    hcc.traceDetail(msg);
  }

  conn.disconnect();
}

The main part of this is the loop that pulls out the identifier for each swept document and adds it to a array. When the loop has completed a valid /run payload is created and the httpPostJson function is called to invoke the REST endpoint.

Batch Job Deep Dive

Now that I’ve covered the overall approach lets focus on the implementation details for the batch job.

I will take each part of the code in turn and describe what its doing. I will also cover the supporting Classes that I created to handle interacting with the MCPTools and watsonx.ai.

Initialisation

Before the script starts to execute several libraries need to be imported and a configuration for a locally run MCP server needs to be defined.

from astrapy.constants import DefaultDocumentType
import asyncio
from astradb import AstraDB
from typing import Any, Optional
    
import sys, os

from mcp_client import MCPHandler
from llm_client import LLMHandler
from astradb import AstraDB
from dotenv import load_dotenv
from datetime import datetime
import json

load_dotenv()

# Load the environment variables
CPBA_USERNAME: str | None=os.getenv('CPBA_USERNAME')
CPBA_PASSWORD: str | None=os.getenv('CPBA_PASSWORD')
CPBA_SERVER_URL: str | None=os.getenv('CPBA_SERVER_URL')
CPBA_CONTENT_STORE: str | None=os.getenv('CPBA_CONTENT_STORE')

# Define the MCP configuration (NB: This could be moved to a config file)
mcp_config: dict[str, dict[str, dict[str, Any]]] = {
      "mcpServers": {
        "core-cs-mcp-server": {
          "command": "uvx",
          "args": [
            "--from",
            "git+https://github.com/ibm-ecm/cs-mcp-server",
            "core-cs-mcp-server"
          ],
          "env": {
            "USERNAME": CPBA_USERNAME,
            "PASSWORD": CPBA_PASSWORD,
            "SERVER_URL": CPBA_SERVER_URL,
            "OBJECT_STORE": CPBA_CONTENT_STORE
          }
        }
      }
    }

The MCP server configuration is defined to pull from the public preview git repository. I addition the necessary environment settings are provided to allow the tools to connect to my FileNet instance.

batch Class

To manage execution of the batch I created a class to handle the necessary processing.


#=====================================================
# batch Class definition
#=====================================================
class batch:

    #=====================================================
    # __init__
    # Constructor for the batch class.
    # parameters:
    #   None
    # returns:
    #   None
    #=====================================================
    def __init__(self) -> None:
        self.mcp_handler: Optional[MCPHandler] = None
        self.llm_handler: Optional[LLMHandler] = None

    #=====================================================
    # process_doc
    # 
    # Executes the steps to process a document and update
    # it in FileNet using MCP tools.
    #
    # parameters:
    #   id: document id
    # returns:
    #   None
    #=====================================================
    async def process_doc(self, id,) -> None:
      # Ensure handlers are initialized
      if self.mcp_handler is None:
          raise RuntimeError("MCP handler not initialized. Call run() first.")
      if self.llm_handler is None:
          raise RuntimeError("LLM handler not initialized. Call run() first.")

      # Step 1: Get the document text extract
      text: Any | None = await self.mcp_handler.get_document_text_extract(id)
      if text is None:
          print("failed to get text extract")
          return
          
      if len(text) > 0:
          print(f'Extracted Text = {text}, {len(text)}')
      else:
          print("Retreived text is empty")
          return

      # Step 2: Determine the document class
      #   call LLMN to identify keywords in the text extract
      keywords: Any | None = self.llm_handler.keywords(input=text)
      print(f'keywords = {keywords}')

      #   call MCP to determine the document class 
      doc_class: Any | None = await self.mcp_handler.determine_class(keywords)

      if doc_class is not None:
        print(f'Determined class = {doc_class['class_description_data']['symbolic_name']}')
      else:
        print("Could not determine document class")
        return

      # Step 3: Update the document class
      #   get the symbolic name for the class and call MCP to update the document to ths class
      symbolic_name = doc_class['class_description_data']['symbolic_name']
      update_doc: Any | None = await self.mcp_handler.update_document_class(id, class_name=symbolic_name)
      if update_doc is not None:
        print("Document class updated")
      else:
        print("Document class was not updated")
        return

      # Step 4: Get the property names for the specific document class
      doc_prop_names: Any | None = await self.mcp_handler.get_class_specific_properties_name(id)
      if doc_prop_names is None:
          print("failed to get document specific property names")
          return  
      else:       
        print(f'Document property names = {doc_prop_names}')

      # Step 5: Get all the properties for the document class
      doc_props: Any | None = await self.mcp_handler.get_class_property_descriptions(class_symbolic_name=symbolic_name)
      if doc_props is None:
          print("failed to get document properties")
          return  

      filtered_props: list[Any] = []

      for prop in doc_props:
        if prop['display_name'] in doc_prop_names:
          filtered_props.append(prop)

      print(f'Filtered properties = {filtered_props}')

      # Step 6: Extract the properties from the document text
      # Create an empty list to hold properties    
      prop_values: list[Any] = []
      # Loop through the properties and try to extract the values
      for prop in filtered_props:
          print(f'Processing property {prop['display_name']}')
          # Set up payload for LLM prompt
          props_json: dict[str, Any] = {
            "name" : prop['display_name'],
            "description" : prop['descriptive_text'],
            "type" : prop['data_type']
          }
          print(f'LLM Parameters = {props_json}')
          # Call LLM to try and extract the property value from the document text
          property_analysis: Any | None = self.llm_handler.property_extractor(input=text,property_details=props_json)
          # If LLM returns a value, process it otherwise skip
          if property_analysis is None:
              continue
          
          # Get the value from the LLM analysis
          analysis_value = property_analysis['value']

          # If the LLM analysis returned a value, process it
          if analysis_value is not None and analysis_value != 'None':
            print(f'Adding value {analysis_value} for {prop['display_name']} to properties array')
            # Derive the symbolc name from the property name
            sym_name = prop['symbolic_name']
            # Check if the property name is a Date (NB: This is a fudge to handle date strings - need to have a better way to indentify properties which are dates)
            if prop['data_type'] == 'DATE':
              date_obj = None
              try:
                  # Parse the date string using strptime with the format 'Year-Month-Day'
                  date_obj = datetime.strptime(analysis_value, "%Y-%m-%d")
              except ValueError:
                try:
                    # Parse the date string using strptime with the format 'Month Day, Year'
                    date_obj = datetime.strptime(analysis_value, "%B %d, %Y")
                except ValueError:
                  try:
                      # Parse the date string using strptime with the format 'Day Month Year'
                      date_obj = datetime.strptime(analysis_value, "%d %B %Y")
                  except ValueError as e:
                      print(f"Invalid date format. Expected format: 'Month Day, Year' (e.g., 'July 1, 2024') or 'Day Month Year' (e.g., '1 July 2024'). Error: {e}")
                      analysis_value = None
              # If date_obj is not None, convert it to a string in the format 'YYYY-MM-DD' and append 'T00:00:00Z' to the end of the string to make ISO format
              if date_obj is not None:
                  analysis_value = date_obj.strftime(format='%Y-%m-%d')
                  analysis_value = analysis_value + "T00:00:00Z"
            # Create a new dictionary with the updated property value
            prop_data: dict[str, Any] = {
              "identifier" : sym_name,
              "value" : analysis_value
            }
            # Append the new property dictionary to the list of property dictionaries
            prop_values.append(prop_data)

      # Create a new dictionary with the updated property value
      new_doc_props: dict[str, Any] = {"properties": prop_values}
      print(f'Property values to set are {new_doc_props}')

      # Step 6: Update the document properties with the new property value
      update_props: Any | None = await self.mcp_handler.update_document_properties(id,document_properties=new_doc_props)
      print(f'Updated {update_props}')

      return

    #=====================================================
    # run
    # 
    # Starts the batch run.
    #
    # Parameters:
    #   batch (list): List of documents to process.
    #   config (dict): Configuration for the batch run.
    # Returns:
    #   None
    # 
    #=====================================================
    async def run(self, batch, config) -> None:

      # Get the LLM API key from the environment variables and create an LLM handler with it.
      api_key: str | None = os.getenv('LLM_API_KEY')
      self.llm_handler = LLMHandler(api_key=api_key) 

      # Create an MCP handler with the configuration.
      self.mcp_handler = MCPHandler(config)
      # Set up the MCP handler.
      await self.mcp_handler.setup_mcp()

      # Process each document in the batch.
      for doc in batch:
        print(f'\n-----------------------doc {doc['identifier']} processing start-----------------------------------\n')
        await self.process_doc(id=doc['identifier'])
        print('\n-----------------------doc processing end-----------------------------------\n')

      #  Close down the MCP handler.
      await self.mcp_handler.close_down()

__init__ function

This is the initialiser for the class and simple creates instance variable for the LLM and MCP handlers.

run function

The run() function manages the initialisation of the supporting LLM and MCP handlers (see later for details about their implementations). These are key to handle the processing of the documents.

Next it works through each document in the batch it was passed and calls the process_doc() function for each.

process_doc function

This is where the actual processing happens. The complete process for a single document is made up of the following steps:

  • Use the get_document_text_extract MCP tool to retrieve the extracted text from FileNet for the specified document identifier.
  • Call LLM Handler to invoke the keywords prompt to identify top three keywords within the extracted text. These keywords are then passed to the determine_class MCP tool to identify the best FileNet class to assign to the document.
  • Update the class of the document to the determined class using the update_document_class MCP tool.
  • Once the document class has been update, the get_class_specific_properties_name MCP tool is called to pull back the names for the specific properties for the class. This is used to build a list of names. The get_class_property_descriptions MCP tool is called next to get the full details for all the properties and this is then filtered using the list of property names created earlier to create a collection of properties to process, each with a full set of details. Each of these properties is then used to populate a request to the LLM property_extractor function. This returns any data that has been found that could populate the property, or None if nothing found.
  • Finally the document properties are updated by calling the update_document_properties MCP tool.

main entry point

This manages the reception of the passed batch identifier, the retrieval of the details from the database. Once this has been done an instance of the batch class is created and this is used to run the batch.

# ========================================================================
# main
#
# Initial entry point for script.  Expects a single string parameter
# which is the ID of the batch to be processed.  The batch ID is used to
# retrieve the documents associated with the batch from the database.
# ========================================================================

if __name__ == "__main__":

  if len(sys.argv) == 1:
     print("Need to be passed a batch ID")
     exit(code=-1)

  # Load Database access details from environment variables.
  token: str | None = os.getenv('ASTRABD_TOKEN')
  endpoint: str | None = os.getenv('ASTRA_ENDPOINT')

  # Create the database hanlder
  astrdb: AstraDB = AstraDB(endpoint=endpoint ,token=token, collection="batch" )

  # Get the batch document from the database
  doc: DefaultDocumentType | None = astrdb.findDoc(doc_id=sys.argv[1])
  if doc is None:
      print(f"No batch found with ID: {sys.argv[1]}")
      exit(code=-1)

  # Get the list of documents associated with the batch
  documents = doc['documents']

  # Create the batch instance and run it
  batch_instance: batch = batch()
  asyncio.run(main=batch_instance.run(batch=documents,config=mcp_config))

MCPHandler class

The MCPHandler class acts as a wrapper for the underlying MCP tools and manages the life cycle of the supporting MCP server.

import asyncio
from mcp_use import MCPClient
import json
import inspect
from typing import Any, List, Optional, Union

#=====================================================
# MCPHandler class
#=====================================================
class MCPHandler:

    #=====================================================
    # __init__
    # Initialize the MCPHandler class
    # Parameters:
    # config (dict): The configuration for the MCPHandler
    # Returns:
    # None
    #=====================================================
    def __init__(self, config) -> None:
        self.client = None 
        self.session = None
        self.config = config

    #=====================================================
    # setup_mcp
    # Setup the MCPClient and create all sessions
    # Parameters:
    # None
    # Returns:
    # None
    #=====================================================
    async def setup_mcp(self) -> None:
        self.client = MCPClient.from_dict(config=self.config)

        await self.client.create_all_sessions()
        self.session = self.client.get_session(server_name="core-cs-mcp-server")

    #=====================================================
    # close_down
    # Close all sessions
    # Parameters:
    # None
    # Returns:
    # None
    #=====================================================
    async def close_down(self) -> None:
        if self.client is not None:
            await self.client.close_all_sessions()

    #=====================================================
    # call_ecm_tool
    # Call the ECM tool
    # Parameters:
    # parameters (dict): The parameters for the ECM tool
    # Returns:
    # Any | None: The result of the ECM tool call
    #=====================================================
    async def __call_ecm_tool(self, parameters) -> Any | None:
        # Check if session is initialized
        if self.session is None:
            print("MCPHanlder Error: MCP session not initialized. Call setup_mcp() first.")
            return None
        
        # Call tool directly
        result: Any = await self.session.call_tool(
            # Inspect pulls out calling method name which should match tool name
            name=inspect.stack()[1][3],
            arguments=parameters
        )

        # Set success flag by negating isError
        if result.isError:
            print(f"MCPHandler error {result}")
            return None
        else:
            # Get data from result
            data: Any = result.content
            # Return both to caller
            return data

    #=====================================================
    # get_document_text_extract
    # Get the text extract for a document tool wrapper
    # Parameters:
    # id (str): The identifier for the document
    # Returns:
    # Any | None: The text extract for the document
    #=====================================================
    async def get_document_text_extract(self, id) -> Any | None:
        # set up MCP tool parameters
        parameters: dict[str, Any] = {
            "identifier": id
        }

        # Call the MCP tool to get the text extract
        content: Any | None = await self.__call_ecm_tool(parameters)
        if content is None:
            return content
        
        # Return the text extract
        return content[0].text

    #=====================================================
    # determine_class
    # Call the ECM tool to determine the class of a document
    # tool wrapper 
    # Parameters:
    # keywords (list[str]): The keywords to use for the classification
    # Returns:
    # Any | None: The classification result
    #=====================================================
    async def determine_class(self, keywords) -> Any | None:
        # set up ECM tool parameters
        parameters: dict[str, Any] = {
            "keywords": keywords,
            "root_class": "Document"
        }

        # Call the ECM tool to get the classification result
        content: Any | None = await self.__call_ecm_tool(parameters)

        if content is None:
            return content
        
        # Return the classification result
        return json.loads(s=content[0].text)
    
    #=====================================================
    # update_document_class
    # Call the ECM tool to update the class of a document
    # tool wrapper 
    # Parameters:
    # id (str): The ID of the document to update
    # class_name (str): The new class name for the document
    # Returns:
    # Any | None: The result of the update
    #=====================================================
    async def update_document_class(self, id, class_name) -> Any | None:
        # set up ECM tool parameters
        parameters: dict[str, Any] = {
            "identifier": id,
            "class_identifier": class_name
        }

        # Call the ECM tool to update the document class
        content: Any | None = await self.__call_ecm_tool(parameters)

        if content is None:
            return content
        
        # Return the update result
        return json.loads(content[0].text)
    
    #=====================================================
    # get_class_specific_properties_name
    # Call the ECM tool to get the class-specific properties
    # of a document tool wrapper 
    # Parameters:
    # id (str): The ID of the document to get properties for
    # Returns:
    # Any | None: The class-specific properties of the document
    #=====================================================
    async def get_class_specific_properties_name(self, id) -> list[str] | None:
        # set up ECM tool parameters
        parameters: dict[str, Any] = {
            "identifier": id
        }

        # Call the ECM tool to get the class-specific properties
        content: Any | None = await self.__call_ecm_tool(parameters)

        if content is None:
            return content
        
        props_names: list[Any] = []
        # Loop through the content and extract the propertiess
        for prop in content:
            props_names.append(prop.text)
  
        # Return the class-specific properties
        return props_names
    
    #=====================================================
    # get_class_specific_properties_name
    # Call the ECM tool to get the class-specific properties
    # of a document tool wrapper 
    # Parameters:
    # id (str): The ID of the document to get properties for
    # Returns:
    # Any | None: The class-specific properties of the document
    #=====================================================
    async def get_class_property_descriptions(self, class_symbolic_name) -> list[Any] | None:
        # set up ECM tool parameters
        parameters: dict[str, Any] = {
            "class_symbolic_name": class_symbolic_name
        }

        # Call the ECM tool to get the class-specific properties
        content: Any | None = await self.__call_ecm_tool(parameters)

        if content is None:
            return content
        
        props: list[Any] = []
        # Loop through the content and extract the propertiess
        for prop in content:
            p = json.loads(s=prop.text)
            props.append(p)

        # Return the class-specific properties
        return props

    #=====================================================
    # update_document_properties
    # Call the ECM tool to update the document properties
    # of a document tool wrapper 
    # Parameters:
    # id (str): The ID of the document to update properties for
    # document_properties (dict[str, Any]): The properties to set for the document
    # Returns:
    # Any | None: The update result
    #=====================================================
    async def update_document_properties(self, id, document_properties=None) -> Any | None:
        # set up ECM tool parameters
        parameters: dict[str, Any] = {
            "identifier": id,
            "document_properties": document_properties
        }
        
        # Call the ECM tool to update the document properties
        content: Any | None = await self.__call_ecm_tool(parameters)

        if content is None:
            return content
        
        # Return the update result
        return json.loads(s=content[0].text)

The setup_mcp() function handles the creation of underlying MCP server and a connection to the server. When all the work has completed close_down() handles shutting down the MCP server sessions.

The rest of the functions act as “shims” for the underlying MCP tools and focus on creating the necessary payloads for the tools and handling the responses. These functions are named the same and the tools they call as when the invoke the __call_ecm_tool()function to make the actual MCP tool call, the name of the calling function is used to define the name of the MCP tool to call.

LLMHandler class

This is responsible for managing the interactions with watsonx.ai. The prompts that are used to support the batch processing are managed in watsonx.ai and deployed to a deployment space so they can be called via REST endpoints.

from requests.models import Response
from typing import Any
import json
import requests


#=====================================================
# LLMHandler class
#=====================================================
class LLMHandler:
    #=====================================================
    # __init__
    # Initialize the LLMHandler class
    # Parameters:
    # api_key (str): The API key for the Watsonx model
    # endpoint (str): The endpoint for the Watsonx model
    # Returns:
    # None
    #=====================================================
    def __init__(self, api_key, endpoint= "https://<watsonx.ai region>/ml/v1/deployments") -> None:

        self.token_url = "https://iam.cloud.ibm.com/identity/token"
        self.prompt_endpoint = endpoint 
        self.api_key = api_key

    #=====================================================
    # keywords
    # Call the Watsonx model to extract keywords
    #
    # Parameters:
    # input (str): The input text to extract keywords from
    # Returns:
    # Any | None: The extracted keywords or None if there is an error
    #=====================================================
    def keywords(self,input) -> Any | None:
        # Populate variables for the prompt
        pVars: dict[str, Any] = {
            "input" : input
        }
        url: str = f"{self.prompt_endpoint}/keywords_v3/text/generation?version=2021-05-01"
        # Generate text by calling the Watsonx deployed prompt
        r: Any | None = self.call_watsonx_prompt(gen_url=url, prompt_variables=pVars)

        # Extract keywords from generated text
        if r is not None:
            gen_text = r[0]['generated_text']
            gen_text = gen_text.strip("```")
            gen_text = gen_text.strip("json")
            keywords = json.loads(gen_text)
            return keywords
        
        # Return keywords
        return None

    #=====================================================
    # property_extractor
    # Call the Watsonx model to extract properties from a given input
    #
    # Parameters:
    # input (str): The input text to extract properties from
    # property_details (dict): The details of the properties to extract
    # Returns:
    # Any | None: The extracted properties or None if there is an error
    #=====================================================
    def property_extractor(self,input, property_details) -> Any | None:
        # Populate variables for the prompt
        pVars: dict[str, Any] = {
            "input" : input,
            "property" : json.dumps(obj=property_details)
        }
        url: str = f"{self.prompt_endpoint}/property_extractor_v3/text/generation?version=2021-05-01"
        # Generate text by calling the Watsonx deployed prompt
        r: Any | None = self.call_watsonx_prompt(gen_url=url, prompt_variables=pVars)
        print(f'Result from watsonx = {r}')     
        if r is not None:
            # Extract keywords from generated text
            gen_text = r[0]['generated_text']

            gen_text = gen_text.strip("```")
            gen_text = gen_text.strip("json")
            gen_text = gen_text.strip("[/EXTRACTED_PROPERTY_SCHEMA]")
            gen_text = gen_text.strip("[EXTRACTED_PROPERTY_SCHEMA]")
            gen_text = gen_text.replace('{{', '{')
            gen_text = gen_text.replace('}}', '}')
            gen_text = gen_text.replace(': None',': null')

            property_result = json.loads(gen_text)
            print(f'Result from watsonx returning {property_result}')
            return property_result 
        
        # Return keywords
        print(f'No result from watsonx returning None')
        return None 

    #=====================================================
    # call_watsonx_prompt
    # 
    # Call Watsonx deployed prompt
    # Parameters:
    #   gen_url: str - URL for generating text
    #   prompt_variables: dict[str, Any] - Prompt variables
    # Returns:
    #   Any | None - Result from Watsonx
    #=====================================================
    def call_watsonx_prompt(self, gen_url, prompt_variables = {}) -> Any | None:
        # Get IAM Token (NB: Could implement a cachine approach here to further optimise)
        # Set up for call
        token_headers: dict[str, str] = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "application/json"
        }
        token_data: dict[str, Any] = {
            "grant_type": "urn:ibm:params:oauth:grant-type:apikey",
            "apikey": self.api_key
        }
        # Call IAM to get token
        token_response: Response = requests.post(url=self.token_url, headers=token_headers, data=token_data)
        if token_response.status_code != 200:
            print(f"response from get token = {token_response.status_code}")
            print(token_response)
            print(f"reason for get token failure = {token_response.reason}")
            # Do hard fail if unable to get token (NB: This could be handled more cleanly but not having access to watsonx impacts batch)
            exit(code=token_response.status_code)

        # Get token from response
        IAM_TOKEN = token_response.json().get("access_token")

        # Call text generation endpoint
        # set up for call
        gen_headers: dict[str, str] = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {IAM_TOKEN}"
        }

        gen_data: dict[str, dict[str, Any]] = {
            "parameters": {
                "prompt_variables": prompt_variables
            }
        }
        # make call
        gen_response: Response = requests.post(url=gen_url, headers=gen_headers, json=gen_data)
        if gen_response.status_code != 200:
            # Call failed so return None
            print(f"response from prompt = {gen_response.status_code}")
            return None

        # Get response from call
        response_json = gen_response.json()

        # Return the results from the call
        return response_json['results']

The class has a utility function call_watsonx_prompt() to handle the technical parts of calling watsonx.ai and two externally called methods

  • keywords() to try and determine the top 3 keywords in the provided text
  • property_extractor() to attempt to extract a value for the provided property from the passed text

The functions define the prompt to call and handle creating the correct variable structure for the prompt. Finally the handle processing the response from the prompt so its processable by the caller function ( process_doc() in this case)

Conclusion

When I started out on this work I did wonder about the overhead of using the MCP layer but all the testing I have done indicates there is very little impact. More importantly using the tools was exceptionally easy and I was able to create this solution in a couple of days (it took a bit longer to iron out bugs in my code though).

Finally the tools provide a consistent abstraction layer for handling this style of batch requirement as well as the more traditional agentic use cases. This means that there is just one thing to learn and all workloads can be addressed.

If you are a FileNet user and you’ve not checked out these MCP tools then I definately recommend doing so.

--

--

Help

Status

About

Careers

Press

Blog

Store

Privacy

Rules

Terms

Text to speech