Building a database query plugin for Semantic Kernel

Most LLM app tutorials stop at a hello-world function. If you want to build database query plugin semantic kernel integration that survives contact with real schemas, you need a native function that introspects tables, executes parameterized reads, and fails closed. This guide walks through a production-minded implementation in Python using the official semantic-kernel package.

Step 1: Scaffold the project and install dependencies

Create a virtual environment and install the Semantic Kernel Python package. As of v1.x the native function API is stable and the @kernel_function decorator is the correct way to expose methods to the planner and chat loop.

python -m venv .venv
source .venv/bin/activate
pip install semantic-kernel python-dotenv

SQLite ships in the standard library, so no extra driver is needed for the example. For Postgres or MySQL, swap the connection string and use the corresponding DBAPI driver, but keep the execution boundary identical.

Step 2: Connect and expose schema metadata

The LLM cannot write valid SQL without knowing table names and column types. Write a small connector that returns schema as a compact string. Avoid dumping sqlite_master verbatim in production—strip index definitions and keep only CREATE TABLE statements.

import sqlite3

class SchemaProvider:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)

    def get_schema(self) -> str:
        cur = self.conn.cursor()
        cur.execute(
            "SELECT name, sql FROM sqlite_master "
            "WHERE type='table' AND name NOT LIKE 'sqlite_%'"
        )
        tables = cur.fetchall()
        return "\n".join(f"-- {name}\n{sql}" for name, sql in tables)

Call get_schema() once at startup and inject it into the system prompt or expose it as a separate @kernel_function. Keeping schema out of the query path avoids repeated introspection overhead on every call.

Step 3: Implement the native query function

To build database query plugin semantic kernel code that is safe, wrap execution in a class with a @kernel_function decorator. The decorator registers the method in the kernel’s function catalog and supplies the description the model uses for tool selection.

from semantic_kernel.functions import kernel_function
import sqlite3

class DatabasePlugin:
    def __init__(self, db_path: str):
        # URI mode enables read-only access at the engine level
        uri = f"file:{db_path}?mode=ro"
        self.conn = sqlite3.connect(uri, uri=True)
        self.conn.row_factory = sqlite3.Row

    @kernel_function(
        name="run_readonly_query",
        description="Execute a read-only SQL SELECT and return rows as a Python list of dicts"
    )
    def run_query(self, query: str) -> str:
        normalized = query.strip().lower()
        if not normalized.startswith("select"):
            return "ERROR: only SELECT queries are permitted"
        try:
            cur = self.conn.cursor()
            cur.execute(query)
            rows = [dict(r) for r in cur.fetchall()]
            return str(rows)
        except Exception as e:
            return f"ERROR: {e}"

The mode=ro URI flag makes SQLite reject writes at the VFS layer. The prefix check is defense in depth, not the primary control. Returning a string keeps the contract simple; the kernel serializes it back to the model as text.

Step 4: Register the plugin and configure the LLM service

Instantiate the kernel, add the plugin, and attach a chat completion service. Semantic Kernel accepts any OpenAI-compatible endpoint. If you want one URL that fronts 240+ models with automatic fallback when a provider is degraded, point base_url at n4n.ai and use your gateway key.

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

async def build_kernel():
    kernel = Kernel()
    kernel.add_plugin(
        DatabasePlugin("example.db"),
        plugin_name="db"
    )
    kernel.add_service(
        OpenAIChatCompletion(
            service_id="chat",
            ai_model_id="gpt-4o-mini",
            api_key="YOUR_KEY",
            base_url="https://api.n4n.ai/v1"  # optional OpenAI-compatible gateway
        )
    )
    return kernel

kernel = asyncio.run(build_kernel())

The plugin registration name (db) becomes the namespace the model sees when it emits a function call. The service_id must match whatever you later request via kernel.get_service().

Step 5: Invoke the plugin directly and via function calling

Direct invocation is the fastest way to confirm wiring before involving the LLM:

async def test_direct():
    func = kernel.get_function("db", "run_readonly_query")
    result = await func.invoke(kernel, {"query": "SELECT 1 AS ok"})
    print(result)

asyncio.run(test_direct())

For autonomous behavior, use a stepwise planner or let the chat service emit tool calls. With OpenAIChatCompletion and a model that supports tools, the kernel routes the call automatically:

async def ask_question(question: str):
    chat = kernel.get_service("chat")
    schema = SchemaProvider("example.db").get_schema()
    prompt = f"Schema:\n{schema}\n\nQuestion: {question}"
    response = await chat.complete_chat_async([{"role": "user", "content": prompt}])
    return response.choices[0].message.content

When you build database query plugin semantic kernel flows with planners, register the schema as a separate @kernel_function so the planner can fetch it on demand instead of stuffing it into every prompt. This reduces token bloat and keeps the planning context clean.

Step 6: Harden against injection and resource abuse

Read-only mode is not enough. A SELECT can still call heavy joins, ATTACH databases, or run subqueries that scan millions of rows. Use SQLite’s authorizer to deny everything except SELECT on the main database:

def _authorizer(action, arg1, arg2, db_name, trigger):
    if action == sqlite3.SQLITE_SELECT and db_name == "main":
        return sqlite3.SQLITE_OK
    return sqlite3.SQLITE_DENY

self.conn.set_authorizer(_authorizer)

Add a row cap by wrapping the query: query = f"SELECT * FROM ({query}) LIMIT 100". Enforce a wall-clock timeout with self.conn.set_progress_handler or run execution in a concurrent.futures.ThreadPoolExecutor with a wait timeout of a few seconds.

Never concatenate user input into the SQL string. The model should emit the full query; if you need filters from end users, pass them as bound parameters via a separate @kernel_function that builds the WHERE clause from typed arguments.

Step 7: Verify end-to-end success

Create a sample database and a test script that exercises the plugin and the LLM path.

sqlite3 example.db "CREATE TABLE users(id INTEGER, name TEXT); INSERT INTO users VALUES(1,'alice');"

async def verify():
    func = kernel.get_function("db", "run_readonly_query")
    out = await func.invoke(kernel, {"query": "SELECT name FROM users"})
    assert "alice" in out, "plugin returned unexpected rows"
    out2 = await func.invoke(kernel, {"query": "DELETE FROM users"})
    assert "ERROR" in out2, "write query was not blocked"
    print("All checks passed")

asyncio.run(verify())

If both assertions pass, the native function works. For the LLM path, send a question like “List all user names” and confirm the kernel logs a function call to db.run_readonly_query and returns alice. Expect output similar to:

[{'name': 'alice'}]

Operational notes

• Cache schema in memory; rebuild only on migration.
• Log every generated query with the calling conversation id for audit and debugging.
• If you use the n4n.ai gateway, provider cache-control hints are forwarded, so repeated identical schema fetches may hit cache and reduce token spend.
• Treat the LLM as an untrusted query author. The patterns above—read-only connections, authorizer hooks, and explicit function schemas—keep that author contained while still letting it answer real questions.

Building a database query plugin semantic kernel style forces you to confront the boundary between natural language and executable SQL. Do it with hard limits, not prompts alone.