This is a guest post written by Houman Kargaran, Engineering Lead at ANZ Bank
The full working code is available at https://github.com/houmanka/PII-Compliant-RAG
In the last blog, we walked through the ingestion Workflow. This one goes a level deeper into a single design decision that does a lot of the compliance heavy lifting: keeping the PII scan, classification, and storage inside one Activity.
That decision looks strange at first. Temporal guidance leans toward small, single-purpose Activities, and the one you'll see in this article is doing three things at once. But it's doing them for a reason.
This client is highly regulated, with strict compliance requirements. Even the Temporal UI runs fully encoded; you need to point your browser at a decoder to read anything, and that is in non-production.
The functional requirements from the previous article set the constraints:
- Application must check for PII with in-house MCP
- Application must classify each complaint with internal ML model
- Application is not allowed to store any PII data
The client has their own PII checker running as an in-house MCP server, and their own ML model for classification. After running both, we persist the result. The scope was not clear on which database, and we could not wait for our business analyst to get back to us, so we had to make an engineering call.
The rest of this article walks through why one Activity is the right shape for this problem, how the storage layer stays swappable, and how heartbeating keeps a long-running streaming Activity honest.
Why all in one Activity?#
You might read the code and notice that the ingestion Activity is very busy. It does the redaction, classification, and data insertion. This is against the very foundation of Temporal.
In every interaction with Temporal, you must remember two very important points. From one Activity to another, you will have a payload size limit. Data passed between Activities will be visible in the Temporal UI, which might be against company policy.
In the functional requirements, we are not allowed to leak data to the Temporal UI even though it is masked.
We use one Activity to stream the data down, feed it to the MCP, classify the redacted text, and store it in a DB. We will pass a file id, which we have stored, to the next Activity.
Local MCP server#
I have simulated this inside the ./localdev/mcp/mcp_pii_server folder. Since this is not a tutorial on how to build an MCP or how it works, I am going to skip it. I am sure you can find lots of resources for it.
How to run it:
fastmcp run ./localdev/mcp/mcp_pii_server/pii_classify.py:mcp --transport http --host 127.0.0.1 --port 8090
After you run the above you will get something like this:
INFO: Started server process [36186]
INFO: Waiting for application startup.
INFO StreamableHTTP session manager started
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8090 (Press CTRL+C to quit)I am using a simple library, presidio-anonymizer, and wrapping it with the MCP. This is not production code, just my attempt to show you how easy it is to connect to a local MCP in Temporal.
Inside an Activity, you will call await session.call_tool("pii_classify", {"text": text})
Check out this Activity: https://github.com/houmanka/PII-Compliant-RAG/blob/main/workflow/activities/ingest_file_activity.py#L47
Local ML classifier#
If you remember, our client has an in-house ML classifier which we need to use. We should not pass any PII to this model, hence I kept this in the same Activity.
While this is one approach, Temporal offers other options. For example, you can use the Claim Check Pattern. The payload codec documentation is worth a read.
For simplicity, I just keep everything in one Activity. The downside is if anything happens, we would be redacting and classifying again. I assume this risk has been evaluated by the risk team and accepted.
I have trained a small classifier model and added it to this project. I have also included the notebook in this repo so you can see how it works. The classifier at our enterprise company is very advanced and trained in-house. Mine is small and I used a Kaggle dataset to do it for this project.
In either case, this is how we call it:
#Load the pre-trained pipeline once at activity start
pipeline = joblib.load(Path(__file__).parents[2] / 'complaints_classifier.joblib')
#Inside the row loop — classifier only ever sees redacted text
predictions = pipeline.predict([redacted_text])
classification = await self.save_classification(name=predictions[0])The classifier is a scikit-learn pipeline loaded from a local .joblib file. It runs entirely on-premise, with no external API call and no data leaving the infrastructure. It only ever receives redacted_text, because the PII scan via MCP has already run at this point. The model never sees raw customer data.
The classification result (billing, fraud, service, etc.) is then persisted to the database alongside the redacted text.
Storage#
The scope was not clear on which database to use: Postgres, Spanner, or something else. In an enterprise, that decision often gets made by a committee and can change. If the Activity called SQLAlchemy or a Postgres driver directly, that change would require touching business logic. That is not acceptable.
Instead, the Activity depends on a DataStore contract, a Python Protocol that defines what storage can do, not how it does it:
class DataStore(Protocol):
def save_complaint(self, complaint: Complaint) -> Complaint: ...
def fetch_unembedded(self, file_id: int) -> list[Complaint]: ...
def mark_embedded(self, file_id: int) -> None: ...
def save_classification(self, name: str) -> Classification: ...
def save_file(self, path: str) -> File: ...
def archive_file(self, file_id: int) -> None: ...The Activity knows nothing about Postgres. It only knows about save_complaint, save_file, and so on. The concrete implementation is wired up in the Worker:
data_storage_provider = build_data_store(kind=DataStorageKind.POSTGRES, config=conf)
ingestion_activity_obj = IngestFileActivity(..., data_store=data_storage_provider, ...)Switching from Postgres to Spanner means changing one line in the Worker; nothing in the Activity code changes.
The same pattern pays off in testing. The Activity accepts any object that satisfies the DataStore protocol, so in tests you pass a fake implementation with no real database required.
In large organisations, this is not premature abstraction. It is a practical defense against the reality that requirements change, DB decisions get revisited, and business logic should not be the casualty of infrastructure debates.
Heartbeating: detecting a stuck Worker#
The Activity streams a CSV row by row. Each row involves an MCP round-trip and a classifier call — a loop that can stall silently if the MCP server hangs or the Worker is killed mid-file.
Without a heartbeat, Temporal cannot distinguish "actively working" from "silently dead." It simply waits for start_to_close_timeout to expire (120 seconds of invisible failure) before retrying.
activity.heartbeat(row_index) sends a liveness signal to the Temporal server after each row completes. On the Workflow side, heartbeat_timeout sets the maximum gap Temporal will tolerate between heartbeats. If no heartbeat arrives within that window, the Activity is considered failed and retried afterwards:
file_id = await workflow.execute_activity(
IngestFileActivity.ingest_file_activity,
FileDetails(path=file_input.path, provider=file_input.provider),
start_to_close_timeout=timedelta(seconds=120),
heartbeat_timeout=timedelta(seconds=20),
)On retry, the Activity reads back the last heartbeat value and skips already-processed rows:
details = activity.info().heartbeat_details
start_row = details[0] + 1 if details else 0
for row_index, line in enumerate(itr):
if row_index < start_row:
continue
# ... process row ...
activity.heartbeat(row_index)A crashed Worker is detected within 20 seconds, and the retry resumes from the row after the last checkpoint, not from the beginning of the file. This is what makes the "no complaint is ever silently lost" promise real for a long-running streaming Activity.
Conclusion#
The single-activity design is not a shortcut. It is a deliberate architectural decision driven by compliance. By keeping the PII scan, classification, and storage inside one Activity, raw complaint text never crosses an Activity boundary and never appears in the Temporal UI. The only thing passed to the next Activity is a file_id. That is the entire security guarantee, enforced by the structure of the Workflow itself.
What's next#
This article covered how Activity isolation enforces the PII boundary at the architectural level. The next article goes deeper into what happens after ingestion: how complaints are embedded using an on-premise model, cached, and pushed to a vector store for semantic search.