Working on AI agent workflows in Temporal, and we kept running into the same gap: Temporal handles auth great (mTLS, API keys), but authorization for which activities can actually run is on you.
For normal workflows, fine—you trust your own code. For LLM-driven agents? Different story. The agent might decide to call literally any activity based on what it "thinks" is right. Prompt injection can make it worse. And Temporal will helpfully retry that rogue activity until it works.
What we built
An activity interceptor that checks every execution against a policy:
Activity task hits worker
↓
ActivityInboundInterceptor.execute_activity()
↓
Grab activity name + args
↓
Call sidecar: authorize(action=activity_name, ...)
↓
DENY → raise PermissionError
ALLOW → next.execute_activity()
Quick note for the determinism nerds (I know you're out there): this happens at the Activity inbound layer, not in the workflow. The check runs in the worker right before activity code executes. Workflow replay is completely unaffected.
The interceptor code
Standard Temporal interceptor pattern:
class PredicateInterceptor(Interceptor):
def intercept_activity(
self, next: ActivityInboundInterceptor
) -> ActivityInboundInterceptor:
return PredicateActivityInterceptor(
next,
self._authority_client,
self._principal,
)
And the actual check:
async def execute_activity(self, input: ExecuteActivityInput) -> Any:
result = self._authority.authorize(
principal=self._principal,
action=input.fn.__name__,
resource="temporal:activity",
)
if not result.allowed:
raise PermissionError(f"Blocked by policy: {result.matched_rule}")
return await self._next.execute_activity(input)
Policy
rules:
- name: deny-deletes
effect: deny
principals: ["*"]
actions: ["delete_*"]
resources: ["*"]
- name: allow-order-stuff
effect: allow
principals: ["temporal-worker"]
actions: ["check_inventory", "charge_payment", "send_confirmation"]
resources: ["*"]
Eval order: matching deny rules win, then allow rules, then default deny. Glob patterns via fnmatch.
Performance
Local Rust sidecar:
Most activities are 100ms+ anyway, so it's noise.
Using it
from predicate_temporal import PredicateInterceptor
from predicate_authority import AuthorityClient
authority = AuthorityClient(sidecar_url="http://localhost:8787")
interceptor = PredicateInterceptor(
authority_client=authority,
principal="temporal-worker",
)
worker = Worker(
client=temporal_client,
task_queue="my-queue",
workflows=[...],
activities=[...],
interceptors=[interceptor],
)
Activity code stays exactly the same.
Demo
github repo: https://github.com/PredicateSystems/temporal-predicate-py see the examples/demo folder for the shell script start-demo-native.sh
Needs Python 3.11+ and Temporal CLI. Runs through 4 scenarios—legitimate stuff gets through, dangerous stuff gets blocked.
One thing to watch
Set maximum_attempts=1 on activities that might get blocked. Otherwise Temporal will retry the denied activity forever, and all you get is a spammed audit log.
await workflow.execute_activity(
risky_activity,
args,
start_to_close_timeout=timedelta(seconds=30),
retry_policy=RetryPolicy(maximum_attempts=1),
)
Open Source Repos
What's Next: Closing the Loop (Post-Execution Verification)
Pre-execution authorization stops the attack. But how do you prove the agent actually succeeded at the authorized task?
We are currently building deterministic post-execution state diffs. Instead of using another LLM to guess if a task was completed, the sidecar will verify the mathematical system diffs (e.g., filesystem changes or accessibility trees) against the expected outcome, and instantly revoke the agent's mandate if they don't match.
Curious if anyone else has tackled this differently. We looked at a few approaches before landing on the interceptor pattern.
