Semantic Kernel Integration

Microsoft Semantic Kernel for agent orchestration.


Components

Component Purpose
Plugins Exposed capabilities
Filters Governance hooks
Memory pgvector integration
Planner Automatic orchestration

Plugins

Define agent capabilities:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function

class SemanticAnalysisPlugin:
    @kernel_function(
        name="analyze_code",
        description="Analyze code semantics"
    )
    async def analyze_code(self, code: str) -> str:
        # Analysis logic
        return f"Analysis of: {code}"
    
    @kernel_function(
        name="generate_artifact",
        description="Generate documentation artifact",
        metadata={"requires_approval": True}  # HITL
    )
    async def generate_artifact(self, spec: str) -> str:
        return f"Artifact for: {spec}"

# Register plugin
kernel = Kernel()
kernel.add_plugin(SemanticAnalysisPlugin(), "semantic")

Filters (Governance)

HITL Filter

1
2
3
4
5
6
7
8
9
from semantic_kernel.filters import FilterTypes

@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def hitl_filter(context, next):
    if context.function.metadata.get("requires_approval"):
        approval = await request_hitl_approval(context)
        if not approval.granted:
            raise HITLRejected(approval.reason)
    return await next(context)

Audit Filter

1
2
3
4
5
6
7
8
9
10
11
12
13
@kernel.filter(FilterTypes.FUNCTION_INVOCATION)
async def audit_filter(context, next):
    start = time.time()
    result = await next(context)
    
    await emit_audit_event({
        "function": context.function.name,
        "input": context.arguments,
        "output": result,
        "duration_ms": (time.time() - start) * 1000
    })
    
    return result

Memory (pgvector)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from semantic_kernel.connectors.memory import PostgresMemoryStore

memory = PostgresMemoryStore(
    connection_string="postgresql://...",
    vector_size=1536
)

kernel.add_memory(memory)

# Store memory
await kernel.memory.save_information(
    collection="project-context",
    text="Project uses spec-first development",
    id="context-001"
)

# Retrieve memory
results = await kernel.memory.search(
    collection="project-context",
    query="development approach",
    limit=5
)

Planner

Automatic function orchestration:

1
2
3
4
5
6
7
8
9
from semantic_kernel.planners import SequentialPlanner

planner = SequentialPlanner(kernel)

plan = await planner.create_plan(
    goal="Analyze code and generate documentation"
)

result = await plan.invoke(kernel)

Mapping to A2A Skills

SK plugins map to Agent Card skills:

1
2
3
4
5
6
7
8
9
10
def generate_agent_card(kernel: Kernel) -> dict:
    skills = []
    for plugin in kernel.plugins:
        for func in plugin.functions:
            skills.append({
                "id": f"{plugin.name}-{func.name}",
                "name": func.name,
                "description": func.description
            })
    return {"skills": skills}

Last Updated: January 2026