agentshield/adapter/
semantic_kernel.rs

1use std::path::Path;
2
3use once_cell::sync::Lazy;
4use regex::Regex;
5
6use crate::config::ScanPathFilter;
7use crate::error::Result;
8use crate::ir::capability::{project_declared_description, project_declared_permissions};
9use crate::ir::taint_builder::build_data_surface;
10use crate::ir::*;
11
12static SK_KERNEL_FN_RE: Lazy<Regex> = Lazy::new(|| {
13    Regex::new(r#"(?s)@kernel_function\s*(?:\(\s*(?:[^\)]*?name\s*=\s*["'`]([^"'`]+)["'`])?(?:[^\)]*?description\s*=\s*["'`]([^"'`]+)["'`])?[^\)]*\))?\s*\n\s*(?:async\s+)?def\s+(\w+)\s*\([^)]*\)\s*(?:->[^:]+)?:\s*(?:\n\s*["'`]([\s\S]*?)["'`])?"#)
14        .expect("static regex pattern is valid")
15});
16
17/// Microsoft Semantic Kernel framework adapter (`semantic-kernel`).
18///
19/// Detects Semantic Kernel projects by looking for:
20/// - `pyproject.toml` / `requirements.txt` with `semantic-kernel`
21/// - Python files importing `from semantic_kernel` or `import semantic_kernel`
22pub struct SemanticKernelAdapter;
23
24impl super::Adapter for SemanticKernelAdapter {
25    fn framework(&self) -> Framework {
26        Framework::SemanticKernel
27    }
28
29    fn detect(&self, root: &Path) -> bool {
30        let pyproject = root.join("pyproject.toml");
31        if pyproject.exists() {
32            if let Some(content) = super::read_file_capped(&pyproject) {
33                if content.contains("semantic-kernel") || content.contains("semantic_kernel") {
34                    return true;
35                }
36            }
37        }
38
39        let requirements = root.join("requirements.txt");
40        if requirements.exists() {
41            if let Some(content) = super::read_file_capped(&requirements) {
42                if content.lines().any(|l| {
43                    let trimmed = l.trim();
44                    trimmed.starts_with("semantic-kernel")
45                        || trimmed.starts_with("semantic_kernel")
46                }) {
47                    return true;
48                }
49            }
50        }
51
52        if super::mcp::has_recursive_python_import(
53            root,
54            &[
55                "from semantic_kernel",
56                "import semantic_kernel",
57                "@kernel_function",
58            ],
59        ) {
60            return true;
61        }
62
63        false
64    }
65
66    fn load(&self, root: &Path, ignore_tests: bool) -> Result<Vec<ScanTarget>> {
67        let filter = ScanPathFilter::for_ignore_tests(ignore_tests);
68        self.load_with_filter(root, &filter)
69    }
70
71    fn load_with_filter(&self, root: &Path, filter: &ScanPathFilter) -> Result<Vec<ScanTarget>> {
72        let name = root
73            .file_name()
74            .map(|n| n.to_string_lossy().to_string())
75            .unwrap_or_else(|| "semantic-kernel-project".into());
76
77        let mut source_files = Vec::new();
78        super::mcp::collect_source_files_with_filter(root, filter, &mut source_files)?;
79
80        source_files.retain(|sf| sf.language == Language::Python);
81
82        let mut tools = Vec::new();
83        for sf in &source_files {
84            tools.extend(extract_semantic_kernel_tools_from_source(&sf.path, &sf.content));
85        }
86
87        for tool in &mut tools {
88            project_declared_permissions(tool);
89            project_declared_description(tool);
90        }
91
92        let execution = super::pipeline::build_execution_surface(&source_files);
93        let data = build_data_surface(&tools, &execution);
94        let dependencies = super::mcp::parse_dependencies(root, filter);
95        let provenance = super::mcp::parse_provenance(root, filter);
96
97        Ok(vec![ScanTarget {
98            name,
99            framework: Framework::SemanticKernel,
100            root_path: root.to_path_buf(),
101            tools,
102            execution,
103            data,
104            dependencies,
105            provenance,
106            source_files,
107        }])
108    }
109}
110
111pub(crate) fn extract_semantic_kernel_tools_from_source(path: &Path, content: &str) -> Vec<ToolSurface> {
112    let mut tools = Vec::new();
113
114    for cap in SK_KERNEL_FN_RE.captures_iter(content) {
115        let explicit_name = cap.get(1).map(|m| m.as_str().to_string());
116        let explicit_desc = cap.get(2).map(|m| m.as_str().to_string());
117        let def_name = cap[3].to_string();
118        let docstring = cap.get(4).map(|m| m.as_str().trim().to_string());
119
120        let tool_name = explicit_name.unwrap_or(def_name);
121        let description = explicit_desc.or(docstring);
122
123        let match_start = cap.get(0).map(|m| m.start()).unwrap_or(0);
124        let line = content[..match_start].lines().count() + 1;
125
126        tools.push(ToolSurface {
127            name: tool_name,
128            description,
129            input_schema: None,
130            output_schema: None,
131            declared_permissions: Vec::new(),
132            declared_capabilities: std::collections::BTreeSet::new(),
133            capability_declarations: Vec::new(),
134            observed_capabilities: std::collections::BTreeSet::new(),
135            capability_evidence: Vec::new(),
136            capability_observation_complete: false,
137            defined_at: Some(SourceLocation {
138                file: path.to_path_buf(),
139                line,
140                column: 0,
141                end_line: Some(line),
142                end_column: None,
143            }),
144        });
145    }
146
147    tools
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use crate::adapter::Adapter;
154
155    #[test]
156    fn detects_semantic_kernel_requirements() {
157        let dir = tempfile::tempdir().unwrap();
158        let req = dir.path().join("requirements.txt");
159        std::fs::write(&req, "semantic-kernel>=1.0.0\n").unwrap();
160
161        let adapter = SemanticKernelAdapter;
162        assert!(adapter.detect(dir.path()));
163    }
164
165    #[test]
166    fn detects_semantic_kernel_import() {
167        let dir = tempfile::tempdir().unwrap();
168        let src = dir.path().join("plugin.py");
169        std::fs::write(&src, "from semantic_kernel.functions import kernel_function\n").unwrap();
170
171        let adapter = SemanticKernelAdapter;
172        assert!(adapter.detect(dir.path()));
173    }
174
175    #[test]
176    fn extracts_semantic_kernel_tools() {
177        let content = r#"
178from semantic_kernel.functions import kernel_function
179
180class DatabasePlugin:
181    @kernel_function(name="query_db", description="Execute a database query")
182    def execute_query(self, sql: str) -> str:
183        """Run SQL query against data warehouse."""
184        return db.query(sql)
185"#;
186        let tools = extract_semantic_kernel_tools_from_source(Path::new("plugin.py"), content);
187        assert_eq!(tools.len(), 1);
188        assert_eq!(tools[0].name, "query_db");
189        assert_eq!(
190            tools[0].description.as_deref(),
191            Some("Execute a database query")
192        );
193    }
194}
agentshield/adapter/semantic_kernel.rs

agentshield/adapter/
semantic_kernel.rs
