Intelligent Command Router using MCP, SQLite3, and WSL
📌 Description:
This project allows you to input natural language queries. Based on the content of the query,
the system decides:
- If it's a SQL-related query → it executes on SQLite3.
- If it's a Linux-related query → it executes on WSL (Linux shell).
It uses:
- MCP (Model Context Protocol) for structured tool execution.
- Hugging Face model to convert NL to Linux and SQL commands.
- FastAPI + FastMCP for API server.
- SQLite3 as database.
- subprocess to interface with WSL.
📁 File Structure:
project/
├── [Link] 🧠 Main MCP Agent with tool router
├── tools/
│ ├── linux_tool.py 🐧 Executes WSL commands
│ └── sql_tool.py Executes SQLite queries
├── db.sqlite3 📂 Your SQLite DB file
└── streamlit_app.py Optional UI (if needed)
📄 tools/linux_tool.py (Line-by-Line Explanation)
This file defines a tool to execute Linux shell commands via WSL.
1. Imports necessary MCP base classes and subprocess.
2. Defines a Tool named `linux_shell` which takes a 'command' string.
3. Uses subprocess to run the command using WSL and captures output.
4. Returns the result or error wrapped in a ToolResponse object.
Code Snippet:
from mcp import Tool, ToolCall, ToolResponse
import subprocess
class LinuxShellTool(Tool):
def get_tool_description(self):
return {
"name": "linux_shell",
"description": "Executes bash commands on a Linux system (WSL)",
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The Linux command to run"
}
},
"required": ["command"]
}
}
def call(self, tool_call: ToolCall) -> ToolResponse:
cmd = tool_call.input["command"]
result = [Link](["wsl", cmd], capture_output=True, text=True, shell=False)
output = [Link]() or [Link]()
return ToolResponse(output=output, tool_call_id=tool_call.id)
📄 tools/sql_tool.py (Line-by-Line Explanation)
This tool runs SQLite3 SQL queries.
1. Connects to SQLite database using `[Link]`.
2. Executes a SQL query provided as input.
3. Fetches results and converts them into readable dictionary format.
4. Returns the output or any error encountered.
Code Snippet:
from mcp import Tool, ToolCall, ToolResponse
import sqlite3
class SQLiteTool(Tool):
def __init__(self, db_path="db.sqlite3"):
self.db_path = db_path
def get_tool_description(self):
return {
"name": "sql_runner",
"description": "Executes SQL queries on the SQLite3 database",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The SQL query to run"
}
},
"required": ["query"]
}
}
def call(self, tool_call: ToolCall) -> ToolResponse:
query = tool_call.input["query"]
try:
conn = [Link](self.db_path)
cursor = [Link]()
[Link](query)
rows = [Link]()
columns = [desc[0] for desc in [Link] or []]
output = "\n".join([str(dict(zip(columns, row))) for row in rows]) if rows else "Query
executed."
[Link]()
[Link]()
except Exception as e:
output = f"Error: {e}"
return ToolResponse(output=output, tool_call_id=tool_call.id)
📄 [Link] (Main Agent)
This is the main file that defines the smart agent logic:
1. Loads a Hugging Face model (`mrm8488/t5-base-finetuned-bash`) to translate NL to
Bash/SQL.
2. Uses keyword matching to detect if a query is SQL or Linux-related.
3. Translates the input and sends it to the appropriate tool (SQL or Linux).
4. Uses FastAPI and FastMCP to expose the agent as a web API.
Code Snippet:
from fastapi import FastAPI
from [Link] import FastMCP
from transformers import pipeline
from tools.linux_tool import LinuxShellTool
from tools.sql_tool import SQLiteTool
model = pipeline("text2text-generation", model="mrm8488/t5-base-finetuned-bash")
def detect_intent(query: str) -> str:
sql_keywords = ["table", "rows", "database", "insert", "select", "from", "where", "column"]
if any(word in [Link]() for word in sql_keywords):
return "sql_runner"
return "linux_shell"
def translate(query: str, intent: str):
return model(query, max_length=64)[0]["generated_text"].strip()
class SmartAgent:
def __init__(self, tools):
[Link] = {tool.get_tool_description()["name"]: tool for tool in tools}
async def ask(self, input: str):
intent = detect_intent(input)
tool = [Link][intent]
translated_input = translate(input, intent)
tool_call = {
"id": "query",
"input": { "command" if intent == "linux_shell" else "query": translated_input }
}
response = [Link](tool_call)
return {
"intent": intent,
"translated_input": translated_input,
"output": [Link]
}
tools = [LinuxShellTool(), SQLiteTool()]
agent = SmartAgent(tools)
app = FastAPI()
app.include_router(FastMCP(agent).router)