From 57805a976690769e0afdd0736c89fc1bbaf12840 Mon Sep 17 00:00:00 2001 From: samueloladji-beep Date: Fri, 29 May 2026 11:40:04 -0700 Subject: [PATCH 1/3] docs: add Vaultak runtime security integration guide and filter sample MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a security integration guide and a runnable Python sample showing how to wire Vaultak into Semantic Kernel using the native filter system (FunctionInvocationFilter + AutoFunctionInvocationFilter). Every plugin call is risk-scored before execution and PII is masked in outputs. - docs/VAULTAK_SECURITY.md — integration guide with quick-start, configuration table, event-coverage table, and links - python/samples/concepts/filtering/vaultak_security_filter.py — end-to-end sample with MathPlugin and TimePlugin Co-Authored-By: Claude Sonnet 4.6 --- docs/VAULTAK_SECURITY.md | 122 ++++++++++++++ .../filtering/vaultak_security_filter.py | 151 ++++++++++++++++++ 2 files changed, 273 insertions(+) create mode 100644 docs/VAULTAK_SECURITY.md create mode 100644 python/samples/concepts/filtering/vaultak_security_filter.py diff --git a/docs/VAULTAK_SECURITY.md b/docs/VAULTAK_SECURITY.md new file mode 100644 index 000000000000..859337d4ea69 --- /dev/null +++ b/docs/VAULTAK_SECURITY.md @@ -0,0 +1,122 @@ +# Vaultak Runtime Security Integration + +[Vaultak](https://vaultak.com) is a runtime security platform that wraps Semantic Kernel agents +with real-time protection. Every plugin function call and LLM tool selection is risk-scored (0–10), +checked against your policy rules, and automatically blocked before it reaches your production +systems — without changing your agent logic. + +## Why use Vaultak with Semantic Kernel? + +SK agents given plugins can cause real damage: deleted records, leaked PII, unauthorized API calls. +Vaultak adds a security layer via SK's native **filter** system: + +- **Risk scoring** on every plugin function call before it executes +- **Policy enforcement** — block calls that violate your rules +- **PII masking** — strip sensitive data from plugin outputs +- **Audit trail** — every call, every score, in your Vaultak dashboard + +## How it works + +Semantic Kernel's [filter system](https://learn.microsoft.com/en-us/semantic-kernel/concepts/enterprise-readiness/filters) +is the native hook for security and observability. Vaultak registers two filters: + +| Filter type | When it runs | Vaultak action | +|---|---|---| +| `FunctionInvocationFilter` | Every `kernel.invoke()` call | Risk-scores the call; raises `KernelFunctionCancelledError` if above threshold; masks PII in output | +| `AutoFunctionInvocationFilter` | Each LLM-selected tool call (auto function calling) | Risk-scores the call; sets `context.terminate = True` to stop the loop if above threshold | + +## Installation + +```bash +pip install vaultak semantic-kernel +``` + +Sign up at [vaultak.com](https://vaultak.com) to get your API key (starts with `vtk_`). + +## Quick start + +```python +import asyncio +import os +from collections.abc import Callable, Coroutine +from typing import Any + +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings +from semantic_kernel.core_plugins import MathPlugin, TimePlugin +from semantic_kernel.exceptions import KernelFunctionCancelledError +from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext +from semantic_kernel.functions import KernelArguments + +from vaultak import Vaultak + +RISK_THRESHOLD = 7.0 +vt = Vaultak(api_key=os.environ["VAULTAK_API_KEY"], agent_name="sk-agent") + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(service_id="chat")) +kernel.add_plugin(MathPlugin(), plugin_name="math") +kernel.add_plugin(TimePlugin(), plugin_name="time") + + +@kernel.filter(FilterTypes.FUNCTION_INVOCATION) +async def vaultak_function_filter( + context: FunctionInvocationContext, + next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + action = f"{context.function.plugin_name}-{context.function.name}" + result = vt.score_action(action=action, context=dict(context.arguments or {})) + if result.score >= RISK_THRESHOLD: + raise KernelFunctionCancelledError( + f"[Vaultak] '{action}' blocked — risk {result.score:.1f}/10" + ) + vt.check_policy(tool_name=action, input_data=str(context.arguments)) + await next(context) + if context.result and context.result.value: + context.result._value = vt.mask_pii(str(context.result.value)) + + +@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) +async def vaultak_auto_filter( + context: AutoFunctionInvocationContext, + next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + action = f"{context.function.plugin_name}-{context.function.name}" + result = vt.score_action(action=action, context={"auto_invoke": "true"}) + if result.score >= RISK_THRESHOLD: + context.terminate = True # Stop auto-invocation loop cleanly + return + await next(context) +``` + +## Configuration + +| Parameter | Type | Default | Description | +|---|---|---|---| +| `api_key` | `str` | — | Your Vaultak API key — required | +| `agent_name` | `str` | `"sk-agent"` | Label for this kernel in the Vaultak dashboard | +| `risk_threshold` | `float` | `7.0` | Score (0–10) above which calls are blocked | +| `verbose` | `bool` | `False` | Log every scored action to stdout | + +## What gets monitored + +| SK event | Vaultak action | +|---|---| +| `kernel.invoke()` (explicit call) | Risk-scores the plugin + function; blocks if above threshold | +| Auto function call selected by LLM | Risk-scores; terminates auto-invocation loop if above threshold | +| Plugin output returned | Scans for PII and masks before the result propagates | +| Policy check fails | Raises exception with dashboard URL | + +## Complete sample + +See [`python/samples/concepts/filtering/vaultak_security_filter.py`](../python/samples/concepts/filtering/vaultak_security_filter.py) +for a runnable end-to-end example with MathPlugin and TimePlugin. + +## Links + +- [Vaultak documentation](https://docs.vaultak.com) +- [PyPI: `vaultak`](https://pypi.org/project/vaultak) +- [GitHub](https://github.com/vaultak/vaultak-python) +- [Dashboard](https://app.vaultak.com) +- [SK Filters concept](https://learn.microsoft.com/en-us/semantic-kernel/concepts/enterprise-readiness/filters) diff --git a/python/samples/concepts/filtering/vaultak_security_filter.py b/python/samples/concepts/filtering/vaultak_security_filter.py new file mode 100644 index 000000000000..cc97208aea57 --- /dev/null +++ b/python/samples/concepts/filtering/vaultak_security_filter.py @@ -0,0 +1,151 @@ +# Copyright (c) Microsoft. All rights reserved. + +""" +Vaultak runtime security filter for Semantic Kernel. + +Demonstrates how to integrate Vaultak (https://vaultak.com) with Semantic Kernel +using two filter types: + +- FunctionInvocationFilter — risk-scores every plugin function call before it + executes and raises BlockedByVaultak when the score exceeds the threshold. +- AutoFunctionInvocationFilter — intercepts each tool call chosen by the LLM + during auto-function-calling, allowing per-call blocking without stopping + the whole invocation loop. + +Install: + pip install vaultak semantic-kernel + +Set environment variables: + OPENAI_API_KEY=sk-... + VAULTAK_API_KEY=vtk_... +""" + +import asyncio +import os +from collections.abc import Callable, Coroutine +from typing import Any + +from semantic_kernel import Kernel +from semantic_kernel.connectors.ai import FunctionChoiceBehavior +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings +from semantic_kernel.contents import ChatHistory +from semantic_kernel.core_plugins import MathPlugin, TimePlugin +from semantic_kernel.exceptions import KernelFunctionCancelledError +from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext +from semantic_kernel.functions import KernelArguments + +from vaultak import Vaultak + +VAULTAK_API_KEY = os.environ.get("VAULTAK_API_KEY", "vtk_...") +RISK_THRESHOLD = 7.0 # Block function calls with a risk score >= this value + +vt = Vaultak(api_key=VAULTAK_API_KEY, agent_name="sk-agent") + +kernel = Kernel() +kernel.add_service(OpenAIChatCompletion(service_id="chat")) +kernel.add_plugin(MathPlugin(), plugin_name="math") +kernel.add_plugin(TimePlugin(), plugin_name="time") + + +# --------------------------------------------------------------------------- +# Filter 1: FunctionInvocationFilter +# Runs around every KernelFunction invocation (including prompt functions). +# Use this to risk-score explicit kernel.invoke() calls. +# --------------------------------------------------------------------------- +@kernel.filter(FilterTypes.FUNCTION_INVOCATION) +async def vaultak_function_filter( + context: FunctionInvocationContext, + next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + """Risk-score every kernel function call before executing it.""" + plugin_name = context.function.plugin_name or "kernel" + function_name = context.function.name + action = f"{plugin_name}-{function_name}" + + # Collect function arguments as context for the risk scorer + args_context = {k: str(v) for k, v in (context.arguments or {}).items()} + + result = vt.score_action(action=action, context=args_context) + + if result.score >= RISK_THRESHOLD: + raise KernelFunctionCancelledError( + f"[Vaultak] Function '{action}' blocked — risk score {result.score:.1f}/10 " + f"exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" + ) + + # Check against configured policy rules + vt.check_policy(tool_name=action, input_data=str(args_context)) + + await next(context) + + # Scan the function output for PII before it propagates + if context.result and context.result.value: + raw_output = str(context.result.value) + context.result._value = vt.mask_pii(raw_output) + + +# --------------------------------------------------------------------------- +# Filter 2: AutoFunctionInvocationFilter +# Runs for each tool call the LLM selects during auto-function-calling. +# Use this to risk-score or block individual tool calls mid-conversation. +# --------------------------------------------------------------------------- +@kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) +async def vaultak_auto_filter( + context: AutoFunctionInvocationContext, + next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]], +) -> None: + """Risk-score each LLM-selected tool call during auto function invocation.""" + plugin_name = context.function.plugin_name or "kernel" + function_name = context.function.name + action = f"{plugin_name}-{function_name}" + + result = vt.score_action(action=action, context={"auto_invoke": "true"}) + + if result.score >= RISK_THRESHOLD: + # Setting terminate=True stops the auto-invocation loop cleanly + # instead of raising an exception that would surface to the user. + context.terminate = True + return + + vt.check_policy(tool_name=action, input_data=action) + await next(context) + + +# --------------------------------------------------------------------------- +# Demo: run a short agent conversation +# --------------------------------------------------------------------------- +execution_settings = OpenAIChatPromptExecutionSettings( + service_id="chat", + max_tokens=1000, + function_choice_behavior=FunctionChoiceBehavior.Auto( + filters={"included_plugins": ["math", "time"]} + ), +) + +history = ChatHistory() +history.add_system_message("You are a helpful assistant. Use math and time tools when asked.") + + +async def chat(user_input: str) -> None: + history.add_user_message(user_input) + arguments = KernelArguments( + settings=execution_settings, + chat_history=history, + user_input=user_input, + ) + result = await kernel.invoke_prompt( + "{{$chat_history}}{{$user_input}}", + arguments=arguments, + ) + print(f"Assistant: {result}") + history.add_assistant_message(str(result)) + + +async def main() -> None: + print("Vaultak security filters active. Every tool call is risk-scored before execution.") + await chat("What is 42 multiplied by 7?") + await chat("What time is it right now?") + + +if __name__ == "__main__": + asyncio.run(main()) From cbac0b03e4eae5c24cd80bdc0ef107e18262d34d Mon Sep 17 00:00:00 2001 From: samueloladji-beep Date: Fri, 29 May 2026 11:45:27 -0700 Subject: [PATCH 2/3] fix: address Copilot review comments on Vaultak security filter - Wrap synchronous vt.score_action(), vt.check_policy(), and vt.mask_pii() calls with asyncio.to_thread() to avoid blocking the event loop - Replace context.result._value (private attr) with the public FunctionResult constructor: context.result = FunctionResult(...) - Update threshold error messages to say "meets or exceeds" to accurately reflect the >= operator - Raise ValueError when VAULTAK_API_KEY env var is missing instead of silently defaulting to a placeholder string - Fix plugin_name None fallback in docs quick start to use `or 'kernel'` consistent with the sample file - Add dashboard URL to exception message in docs quick start snippet Co-Authored-By: Claude Sonnet 4.6 --- docs/VAULTAK_SECURITY.md | 29 +++++++++++------ .../filtering/vaultak_security_filter.py | 31 ++++++++++++------- 2 files changed, 39 insertions(+), 21 deletions(-) diff --git a/docs/VAULTAK_SECURITY.md b/docs/VAULTAK_SECURITY.md index 859337d4ea69..720723b23ebb 100644 --- a/docs/VAULTAK_SECURITY.md +++ b/docs/VAULTAK_SECURITY.md @@ -22,7 +22,7 @@ is the native hook for security and observability. Vaultak registers two filters | Filter type | When it runs | Vaultak action | |---|---|---| -| `FunctionInvocationFilter` | Every `kernel.invoke()` call | Risk-scores the call; raises `KernelFunctionCancelledError` if above threshold; masks PII in output | +| `FunctionInvocationFilter` | Every `kernel.invoke()` call | Risk-scores the call; raises `KernelFunctionCancelledError` if score meets or exceeds threshold; masks PII in output | | `AutoFunctionInvocationFilter` | Each LLM-selected tool call (auto function calling) | Risk-scores the call; sets `context.terminate = True` to stop the loop if above threshold | ## Installation @@ -47,12 +47,19 @@ from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAICh from semantic_kernel.core_plugins import MathPlugin, TimePlugin from semantic_kernel.exceptions import KernelFunctionCancelledError from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext -from semantic_kernel.functions import KernelArguments +from semantic_kernel.functions import FunctionResult, KernelArguments from vaultak import Vaultak +_api_key = os.environ.get("VAULTAK_API_KEY") +if not _api_key: + raise ValueError( + "VAULTAK_API_KEY environment variable is not set. " + "Sign up at https://vaultak.com to get your API key." + ) + RISK_THRESHOLD = 7.0 -vt = Vaultak(api_key=os.environ["VAULTAK_API_KEY"], agent_name="sk-agent") +vt = Vaultak(api_key=_api_key, agent_name="sk-agent") kernel = Kernel() kernel.add_service(OpenAIChatCompletion(service_id="chat")) @@ -65,16 +72,18 @@ async def vaultak_function_filter( context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], ) -> None: - action = f"{context.function.plugin_name}-{context.function.name}" - result = vt.score_action(action=action, context=dict(context.arguments or {})) + action = f"{context.function.plugin_name or 'kernel'}-{context.function.name}" + result = await asyncio.to_thread(vt.score_action, action=action, context=dict(context.arguments or {})) if result.score >= RISK_THRESHOLD: raise KernelFunctionCancelledError( - f"[Vaultak] '{action}' blocked — risk {result.score:.1f}/10" + f"[Vaultak] '{action}' blocked — risk {result.score:.1f}/10 " + f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" ) - vt.check_policy(tool_name=action, input_data=str(context.arguments)) + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=str(context.arguments)) await next(context) if context.result and context.result.value: - context.result._value = vt.mask_pii(str(context.result.value)) + masked = await asyncio.to_thread(vt.mask_pii, str(context.result.value)) + context.result = FunctionResult(function=context.function, value=masked) @kernel.filter(FilterTypes.AUTO_FUNCTION_INVOCATION) @@ -82,8 +91,8 @@ async def vaultak_auto_filter( context: AutoFunctionInvocationContext, next: Callable[[AutoFunctionInvocationContext], Coroutine[Any, Any, None]], ) -> None: - action = f"{context.function.plugin_name}-{context.function.name}" - result = vt.score_action(action=action, context={"auto_invoke": "true"}) + action = f"{context.function.plugin_name or 'kernel'}-{context.function.name}" + result = await asyncio.to_thread(vt.score_action, action=action, context={"auto_invoke": "true"}) if result.score >= RISK_THRESHOLD: context.terminate = True # Stop auto-invocation loop cleanly return diff --git a/python/samples/concepts/filtering/vaultak_security_filter.py b/python/samples/concepts/filtering/vaultak_security_filter.py index cc97208aea57..277ed70510c6 100644 --- a/python/samples/concepts/filtering/vaultak_security_filter.py +++ b/python/samples/concepts/filtering/vaultak_security_filter.py @@ -7,7 +7,8 @@ using two filter types: - FunctionInvocationFilter — risk-scores every plugin function call before it - executes and raises BlockedByVaultak when the score exceeds the threshold. + executes and raises KernelFunctionCancelledError when the score meets or + exceeds the threshold. - AutoFunctionInvocationFilter — intercepts each tool call chosen by the LLM during auto-function-calling, allowing per-call blocking without stopping the whole invocation loop. @@ -32,14 +33,20 @@ from semantic_kernel.core_plugins import MathPlugin, TimePlugin from semantic_kernel.exceptions import KernelFunctionCancelledError from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext -from semantic_kernel.functions import KernelArguments +from semantic_kernel.functions import FunctionResult, KernelArguments from vaultak import Vaultak -VAULTAK_API_KEY = os.environ.get("VAULTAK_API_KEY", "vtk_...") -RISK_THRESHOLD = 7.0 # Block function calls with a risk score >= this value +_api_key = os.environ.get("VAULTAK_API_KEY") +if not _api_key: + raise ValueError( + "VAULTAK_API_KEY environment variable is not set. " + "Sign up at https://vaultak.com to get your API key." + ) + +RISK_THRESHOLD = 7.0 # Block function calls with a risk score that meets or exceeds this value -vt = Vaultak(api_key=VAULTAK_API_KEY, agent_name="sk-agent") +vt = Vaultak(api_key=_api_key, agent_name="sk-agent") kernel = Kernel() kernel.add_service(OpenAIChatCompletion(service_id="chat")) @@ -65,23 +72,25 @@ async def vaultak_function_filter( # Collect function arguments as context for the risk scorer args_context = {k: str(v) for k, v in (context.arguments or {}).items()} - result = vt.score_action(action=action, context=args_context) + # Wrap synchronous SDK calls with asyncio.to_thread to avoid blocking the event loop + result = await asyncio.to_thread(vt.score_action, action=action, context=args_context) if result.score >= RISK_THRESHOLD: raise KernelFunctionCancelledError( f"[Vaultak] Function '{action}' blocked — risk score {result.score:.1f}/10 " - f"exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" + f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" ) # Check against configured policy rules - vt.check_policy(tool_name=action, input_data=str(args_context)) + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=str(args_context)) await next(context) # Scan the function output for PII before it propagates if context.result and context.result.value: raw_output = str(context.result.value) - context.result._value = vt.mask_pii(raw_output) + masked = await asyncio.to_thread(vt.mask_pii, raw_output) + context.result = FunctionResult(function=context.function, value=masked) # --------------------------------------------------------------------------- @@ -99,7 +108,7 @@ async def vaultak_auto_filter( function_name = context.function.name action = f"{plugin_name}-{function_name}" - result = vt.score_action(action=action, context={"auto_invoke": "true"}) + result = await asyncio.to_thread(vt.score_action, action=action, context={"auto_invoke": "true"}) if result.score >= RISK_THRESHOLD: # Setting terminate=True stops the auto-invocation loop cleanly @@ -107,7 +116,7 @@ async def vaultak_auto_filter( context.terminate = True return - vt.check_policy(tool_name=action, input_data=action) + await asyncio.to_thread(vt.check_policy, tool_name=action, input_data=action) await next(context) From 8a0d89c09ab4e2fdbfb17341e2612ffe40f3a246 Mon Sep 17 00:00:00 2001 From: samueloladji-beep Date: Fri, 29 May 2026 14:28:23 -0700 Subject: [PATCH 3/3] Fix SK filter: use OperationCancelledException, skip prompt functions - Replace KernelFunctionCancelledError (doesn't exist) with OperationCancelledException - Add plugin_name is None guard to FunctionInvocationFilter so only real plugin/tool calls are risk-scored, not inline prompt invocations Co-Authored-By: Claude Sonnet 4.6 --- docs/VAULTAK_SECURITY.md | 12 ++++++++---- .../concepts/filtering/vaultak_security_filter.py | 13 +++++++++---- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/docs/VAULTAK_SECURITY.md b/docs/VAULTAK_SECURITY.md index 720723b23ebb..77e260ead04a 100644 --- a/docs/VAULTAK_SECURITY.md +++ b/docs/VAULTAK_SECURITY.md @@ -22,7 +22,7 @@ is the native hook for security and observability. Vaultak registers two filters | Filter type | When it runs | Vaultak action | |---|---|---| -| `FunctionInvocationFilter` | Every `kernel.invoke()` call | Risk-scores the call; raises `KernelFunctionCancelledError` if score meets or exceeds threshold; masks PII in output | +| `FunctionInvocationFilter` | Every `kernel.invoke()` call (plugin functions only) | Risk-scores the call; raises `OperationCancelledException` if score meets or exceeds threshold; masks PII in output | | `AutoFunctionInvocationFilter` | Each LLM-selected tool call (auto function calling) | Risk-scores the call; sets `context.terminate = True` to stop the loop if above threshold | ## Installation @@ -45,7 +45,7 @@ from semantic_kernel import Kernel from semantic_kernel.connectors.ai import FunctionChoiceBehavior from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings from semantic_kernel.core_plugins import MathPlugin, TimePlugin -from semantic_kernel.exceptions import KernelFunctionCancelledError +from semantic_kernel.exceptions import OperationCancelledException from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext from semantic_kernel.functions import FunctionResult, KernelArguments @@ -72,10 +72,14 @@ async def vaultak_function_filter( context: FunctionInvocationContext, next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], ) -> None: - action = f"{context.function.plugin_name or 'kernel'}-{context.function.name}" + # Skip inline prompt functions — only score real plugin calls + if context.function.plugin_name is None: + await next(context) + return + action = f"{context.function.plugin_name}-{context.function.name}" result = await asyncio.to_thread(vt.score_action, action=action, context=dict(context.arguments or {})) if result.score >= RISK_THRESHOLD: - raise KernelFunctionCancelledError( + raise OperationCancelledException( f"[Vaultak] '{action}' blocked — risk {result.score:.1f}/10 " f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" ) diff --git a/python/samples/concepts/filtering/vaultak_security_filter.py b/python/samples/concepts/filtering/vaultak_security_filter.py index 277ed70510c6..f6db9df30b22 100644 --- a/python/samples/concepts/filtering/vaultak_security_filter.py +++ b/python/samples/concepts/filtering/vaultak_security_filter.py @@ -7,7 +7,7 @@ using two filter types: - FunctionInvocationFilter — risk-scores every plugin function call before it - executes and raises KernelFunctionCancelledError when the score meets or + executes and raises OperationCancelledException when the score meets or exceeds the threshold. - AutoFunctionInvocationFilter — intercepts each tool call chosen by the LLM during auto-function-calling, allowing per-call blocking without stopping @@ -31,7 +31,7 @@ from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion, OpenAIChatPromptExecutionSettings from semantic_kernel.contents import ChatHistory from semantic_kernel.core_plugins import MathPlugin, TimePlugin -from semantic_kernel.exceptions import KernelFunctionCancelledError +from semantic_kernel.exceptions import OperationCancelledException from semantic_kernel.filters import AutoFunctionInvocationContext, FilterTypes, FunctionInvocationContext from semantic_kernel.functions import FunctionResult, KernelArguments @@ -65,7 +65,12 @@ async def vaultak_function_filter( next: Callable[[FunctionInvocationContext], Coroutine[Any, Any, None]], ) -> None: """Risk-score every kernel function call before executing it.""" - plugin_name = context.function.plugin_name or "kernel" + # Skip prompt functions (plugin_name is None for inline prompts) + if context.function.plugin_name is None: + await next(context) + return + + plugin_name = context.function.plugin_name function_name = context.function.name action = f"{plugin_name}-{function_name}" @@ -76,7 +81,7 @@ async def vaultak_function_filter( result = await asyncio.to_thread(vt.score_action, action=action, context=args_context) if result.score >= RISK_THRESHOLD: - raise KernelFunctionCancelledError( + raise OperationCancelledException( f"[Vaultak] Function '{action}' blocked — risk score {result.score:.1f}/10 " f"meets or exceeds threshold {RISK_THRESHOLD}. Review at app.vaultak.com" )