Tool guard

pyproject.toml

@@ -56,6 +56,7 @@ dependencies = [
     "psycopg[binary]>=3.2",
     "cuga-oak-health; python_version>='3.12'",
     "aiosmtpd",
+    "toolguard>=0.2.17",
 ]
 
 [project.optional-dependencies]

src/cuga/backend/cuga_graph/policy/filesystem_sync.py

@@ -134,6 +134,12 @@ def _policy_to_markdown(self, policy: Policy) -> str:
             if policy.target_apps:
                 frontmatter['target_apps'] = policy.target_apps
             frontmatter['prepend'] = policy.prepend
+            if policy.tool_guards:
+                # Convert ToolGuard objects to dict for YAML serialization
+                frontmatter['tool_guards'] = {
+                    tool_name: guard.model_dump()
+                    for tool_name, guard in policy.tool_guards.items()
+                }
             content = policy.guide_content or ""
         elif isinstance(policy, IntentGuard):
             if policy.response:

src/cuga/backend/cuga_graph/policy/folder_loader.py

@@ -210,6 +210,16 @@ def create_tool_guide_from_markdown(
     if not triggers:
         triggers = [AlwaysTrigger()]
 
+    # Parse tool_guards if present
+    tool_guards = None
+    tool_guards_data = frontmatter.get('tool_guards')
+    if tool_guards_data and isinstance(tool_guards_data, dict):
+        from cuga.backend.cuga_graph.policy.models import ToolGuard
+        tool_guards = {
+            tool_name: ToolGuard(**guard_data)
+            for tool_name, guard_data in tool_guards_data.items()
+        }
+
     return ToolGuide(
         id=frontmatter.get('id', f"tool_guide_{Path(file_path).stem}"),
         name=name,
@@ -219,6 +229,7 @@ def create_tool_guide_from_markdown(
         target_apps=frontmatter.get('target_apps'),
         guide_content=content,
         prepend=frontmatter.get('prepend', False),
+        tool_guards=tool_guards,
         priority=frontmatter.get('priority', 50),
         enabled=frontmatter.get('enabled', True),
     )

src/cuga/backend/cuga_graph/policy/models.py

@@ -201,6 +201,26 @@ def validate_trigger_targets(self):
         return self
 
 
+class ToolGuard(BaseModel):
+    """Guard configuration for a specific tool with compliance rules."""
+
+    violating_examples: List[str] = Field(
+        default_factory=list, description="Examples of violating usage patterns"
+    )
+    compliance_examples: List[str] = Field(
+        default_factory=list, description="Examples of compliant usage patterns"
+    )
+    policy_code: str = Field(
+        default="",
+        description=(
+            "Python code that validates tool usage compliance. "
+            "This code is executed in a sandboxed environment using the toolguard library. "
+            "Only trusted administrators with manage access should be allowed to modify policy code. "
+            "While sandboxed, policy code should still be reviewed for correctness and performance."
+        )
+    )
+
+
 class ToolGuide(BaseModel):
     """Policy that enriches tool descriptions with additional markdown content."""
 
@@ -215,6 +235,9 @@ class ToolGuide(BaseModel):
     )
     guide_content: str = Field(..., description="Markdown content to append to tool descriptions")
     prepend: bool = Field(False, description="Whether to prepend content instead of appending")
+    tool_guards: Optional[Dict[str, ToolGuard]] = Field(
+        default=None, description="Optional guard configurations per tool (key: tool_name, value: ToolGuard)"
+    )
     metadata: Dict[str, Any] = Field(default_factory=dict, description="Additional metadata")
     priority: int = Field(0, description="Priority when multiple guides match (higher = more important)")
     enabled: bool = Field(True, description="Whether this guide is active")

src/cuga/backend/cuga_graph/policy/storage.py

@@ -177,6 +177,14 @@ async def _generate_policy_embedding(self, policy: Policy) -> List[float]:
                     text_parts.append(policy.guide_content[:300])
                 if policy.target_tools and "*" not in policy.target_tools:
                     text_parts.append(f"Tools: {', '.join(policy.target_tools[:10])}")
+                if policy.tool_guards:
+                    # Add tool guard information to search text
+                    for tool_name, guard in policy.tool_guards.items():
+                        text_parts.append(f"Guard for {tool_name}")
+                        if guard.violating_examples:
+                            text_parts.append(f"Violations: {' '.join(guard.violating_examples[:3])}")
+                        if guard.compliance_examples:
+                            text_parts.append(f"Compliance: {' '.join(guard.compliance_examples[:3])}")
 
             elif isinstance(policy, OutputFormatter):
                 # OutputFormatter-specific content

src/cuga/backend/cuga_graph/policy/tool_guard/tests/__init__.py

@@ -0,0 +1,3 @@
+"""Tests for tool guard policies."""
+
+