← Writing a plugin
Extend kcode
Find issues in a local SQLite database
Connect a local MCP server that lets the agent find issues by status and title. The Python server uses read-only, parameterized queries. It exposes no arbitrary SQL or database-write tool.
The database builder supplies three sample issues. The server implements tool discovery and queries over MCP stdio; .mcp.json tells kcode how to launch it. Tests drive both the Python protocol and kcode’s MCP runtime, including registration, query results, and unload.
sqlite-issues/README.md
# Query a local issue database through MCP
This example runs a Python stdio MCP server, not an in-process Lua or Wasm plugin. The agent calls `issues__find_issues` to find sample issues by status and title text. It cannot supply SQL or modify the database.
## Set up
Requires Python 3 with SQLite support. Save `server.py` and `create_sample.py` together, then create a new sample database:
```sh
python3 -B create_sample.py /absolute/path/to/issues.db
python3 -B test_server.py
```
Copy the bundled `.mcp.json` to your project root and replace both absolute paths with your real paths. If the project already has `.mcp.json`, merge the `issues` entry into its `mcpServers` object rather than replacing other servers. Trusted project configuration is required. Start a new session to connect the server.
Ask the agent to find open issues mentioning SVG. The registered tool takes:
```json
{"status":"open","text":"SVG"}
```
The sample result is issue 2, “Add SVG export”. Results contain issue IDs, titles and status, ordered by ID and capped at 50 rows. Omit either filter to broaden the query. Matching is literal substring matching, not SQL patterns.
## How the files cooperate
- `create_sample.py` creates a new SQLite file with three demonstration issues and refuses overwrite
- `server.py` implements MCP initialization, tool discovery and invocation over newline-delimited JSON-RPC on stdin/stdout
- `.mcp.json` tells kcode which local process to launch and where the database is
- `test_server.py` drives the actual server subprocess and verifies query results, invalid arguments, literal SQL-like input and unchanged database bytes
The server opens SQLite in read-only mode, enables query-only mode, uses bound parameters and exposes no arbitrary SQL tool. Its stdout is reserved for protocol messages. This is a local demonstration server, not an authenticated network service.
## Check through kcode
The kcode repository contains an opt-in integration test for connection, tool registration, query invocation and unload:
```sh
KCODE_SQLITE_MCP_DIR=/absolute/path/to/sqlite-issues cargo test -p plugin-runtime-mcp --test sqlite_issues -- --ignored
```sqlite-issues/.mcp.json
{
"mcpServers": {
"issues": {
"command": "python3",
"args": [
"-B",
"/absolute/path/to/sqlite-issues/server.py",
"--database",
"/absolute/path/to/issues.db"
]
}
}
}sqlite-issues/create_sample.py
"""Create a new demonstration database, never overwrite an existing file."""
from contextlib import closing
import sqlite3
import sys
from pathlib import Path
def create(path):
path = Path(path)
with path.open('xb'):
pass
try:
with closing(sqlite3.connect(path)) as conn:
conn.execute("CREATE TABLE issues (id INTEGER PRIMARY KEY, title TEXT NOT NULL, status TEXT NOT NULL CHECK(status IN ('open','closed')))")
conn.executemany('INSERT INTO issues VALUES (?, ?, ?)', [(1, 'Fix mobile menu focus', 'open'), (2, 'Add SVG export', 'open'), (3, 'Update installation guide', 'closed')])
conn.commit()
except BaseException:
path.unlink()
raise
if __name__ == '__main__':
create(sys.argv[1])sqlite-issues/server.py
"""Local MCP stdio server for a read-only sample issue database."""
import argparse
import json
from pathlib import Path
from contextlib import closing
import sqlite3
import sys
TOOLS = [{'name': 'find_issues', 'description': 'Find up to 50 issues by status and title text in the local sample database', 'inputSchema': {'type': 'object', 'properties': {'status': {'type': 'string', 'enum': ['open', 'closed']}, 'text': {'type': 'string', 'maxLength': 200}}, 'additionalProperties': False}}]
def find_issues(database, args):
if not isinstance(args, dict) or set(args) - {'status', 'text'}:
raise ValueError('arguments must contain only status and text')
status, text = args.get('status'), args.get('text', '')
if status is not None and status not in ['open', 'closed']:
raise ValueError('status must be open or closed')
if not isinstance(text, str) or len(text) > 200:
raise ValueError('text must be at most 200 characters')
uri = Path(database).resolve(strict=True).as_uri() + '?mode=ro'
with closing(sqlite3.connect(uri, uri=True, timeout=2)) as conn:
conn.execute('PRAGMA query_only=ON')
rows = conn.execute('SELECT id, title, status FROM issues WHERE (? IS NULL OR status = ?) AND instr(lower(title), lower(?)) > 0 ORDER BY id LIMIT 50', (status, status, text)).fetchall()
return [{'id': row[0], 'title': row[1], 'status': row[2]} for row in rows]
def handle(database, request):
if not isinstance(request, dict) or request.get('jsonrpc') != '2.0':
return {'jsonrpc': '2.0', 'id': None, 'error': {'code': -32600, 'message': 'Invalid request'}}
if 'id' not in request:
return None
response = {'jsonrpc': '2.0', 'id': request['id']}
method, params = request.get('method'), request.get('params', {})
if method == 'initialize':
response['result'] = {'protocolVersion': '2024-11-05', 'capabilities': {'tools': {}}, 'serverInfo': {'name': 'sqlite-issues', 'version': '0.1.0'}}
elif method == 'ping':
response['result'] = {}
elif method == 'tools/list':
response['result'] = {'tools': TOOLS}
elif method == 'tools/call':
try:
if not isinstance(params, dict) or params.get('name') != 'find_issues':
raise ValueError('unknown tool')
rows = find_issues(database, params.get('arguments', {}))
response['result'] = {'content': [{'type': 'text', 'text': json.dumps(rows)}], 'isError': False}
except (ValueError, OSError, sqlite3.Error) as error:
response['result'] = {'content': [{'type': 'text', 'text': str(error)}], 'isError': True}
else:
response['error'] = {'code': -32601, 'message': 'Method not found'}
return response
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--database', required=True)
args = parser.parse_args()
for line in sys.stdin:
try:
request = json.loads(line)
response = handle(args.database, request)
except (ValueError, TypeError) as error:
response = {'jsonrpc': '2.0', 'id': None, 'error': {'code': -32700, 'message': str(error)}}
if response is not None:
print(json.dumps(response), flush=True)
if __name__ == '__main__':
main()sqlite-issues/test_server.py
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from create_sample import create
from server import find_issues
class IssuesTests(unittest.TestCase):
def test_stdio_and_read_only_queries(self):
with tempfile.TemporaryDirectory() as temp:
db = Path(temp) / 'issues.db'
create(db)
before = db.read_bytes()
requests = [
{'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2024-11-05', 'capabilities': {}, 'clientInfo': {'name': 'test', 'version': '1'}}},
{'jsonrpc': '2.0', 'method': 'notifications/initialized'},
{'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list'},
{'jsonrpc': '2.0', 'id': 3, 'method': 'tools/call', 'params': {'name': 'find_issues', 'arguments': {'status': 'open', 'text': 'SVG'}}},
{'jsonrpc': '2.0', 'id': 4, 'method': 'tools/call', 'params': {'name': 'find_issues', 'arguments': {'sql': 'DROP TABLE issues'}}},
]
result = subprocess.run([sys.executable, '-B', str(Path(__file__).with_name('server.py')), '--database', str(db)], input='\n'.join(json.dumps(r) for r in requests) + '\n', text=True, capture_output=True, timeout=10, check=True)
replies = [json.loads(line) for line in result.stdout.splitlines()]
self.assertEqual([r['id'] for r in replies], [1, 2, 3, 4])
self.assertEqual(replies[1]['result']['tools'][0]['name'], 'find_issues')
self.assertEqual(json.loads(replies[2]['result']['content'][0]['text']), [{'id': 2, 'title': 'Add SVG export', 'status': 'open'}])
self.assertTrue(replies[3]['result']['isError'])
self.assertEqual(find_issues(db, {'text': "' OR 1=1 --"}), [])
self.assertEqual(db.read_bytes(), before)
with self.assertRaises(FileExistsError):
create(db)
if __name__ == '__main__':
unittest.main()