Command-based MCP Server – n8n Nodes

Community nodes that let n8n talk to a command-based Model Context Protocol (MCP) server exposed over TCP. Use them to run parameterised MSSQL queries from any workflow via the MCP bridge—perfect for LAN setups such as 192.168.10.122 → 192.168.10.25.

Author: Abdurrahim Ceylan ([email protected])

Features

• Connects directly to a long-running MCP server via TCP (default 5555).
• Stores MCP Host/Port + MSSQL Server/DB/User/Password + Timeout inside a reusable n8n credential so you only enter them once.
• Streams the MSSQL rowset back to n8n as JSON or as a table (one item per row), optionally attaching the raw MCP payload.
• Ships with a reference MCP Python server (server/command_based_mcp_server.py) and its tool manifest (mcp.json) in case you want to run your own host.
• Provides both a classic node and an AI Agent Tool node (MCP SQL Tool) so n8n’s AI Agent can call the MCP server autonomously.

Installation

git clone https://github.com/rahimceylan/n8n-nodes-command-based-mcp.git
cd n8n-nodes-command-based-mcp
npm install
npm run build

Then copy/symlink the folder into your n8n instance under ~/.n8n/custom/ (or use the N8N_CUSTOM_EXTENSIONS env var) and restart n8n.

Configuration

1. Run (or reuse) an MCP Server

   • Start the bundled TCP server (defaults to 0.0.0.0:5555):

     # Windows PowerShell
     setx MCP_HOST 0.0.0.0
     setx MCP_PORT 5555
     python server/command_based_mcp_server.py

   • Or on Linux/macOS:

     export MCP_HOST=0.0.0.0
     export MCP_PORT=5555
     python server/command_based_mcp_server.py

   • Ensure the host has pyodbc and the Microsoft ODBC Driver 17 for SQL Server installed:

     pip install pyodbc

2. Create the Credential Once

   • In n8n, go to Settings → Credentials and add Command-based MCP Connection.
   • Provide:
     • MCP Server Host: e.g. 192.168.11.222
     • MCP Server Port: e.g. 5555
     • MSSQL Server Address: e.g. 192.168.10.25
     • Database Name: e.g. kacmaz2025
     • Username: e.g. rapor
     • Password: e.g. ••••••
     • Timeout (seconds): optional, defaults to 60

3. Configure the Node

   • Drag the Command-based MCP Server node into a workflow.
   • Select the credential you just created.
   • Fill only the per-query fields:
     • Result Format: JSON (Single Item) or Table (One Item Per Row)
     • Result Field Name (visible only when JSON is selected; default rows)
     • Include Raw Response toggle
     • SQL Query: e.g. SELECT * FROM [dbo].[TMPMALIYET]

4. Expose It as an AI Agent Tool (Optional)

   • Drag the MCP SQL Tool node from the Tools panel (under Model Context Protocol).

• Point it to the same credentials, optionally provide a Default SQL Query, and describe when the agent should run it.
• The tool exposes a single structured parameter sqlQuery; the agent will pass the statement (or fall back to the default) and receives either JSON ({sql, rows}) or a Markdown table depending on the Result Format you set.

5. Run the Workflow

   • The node opens a TCP socket to the MCP host, sends a JSON payload like:

     {"id":"123","method":"call_tool","params":{"name":"query_mssql","arguments":{"server":"192.168.10.25","database":"kcm25","username":"Username","password":"prpr123","sql":"SELECT TOP 10 * FROM STOKKART"}}}

   • Your MCP server executes the query and replies with the rowset under result.content[0].data.

Development

• npm run lint – Lints all TypeScript sources.
• npm run build – Compiles TypeScript to dist.
• npm test – Alias for npm run lint.

Publish to npm once validated:

npm login --scope=@rahimceylan
npm version patch
npm publish --access public

MCP Transport Overview

The node spawns the Python script in a child process and sends a single JSON line shaped like:

{
  "id": "123",
  "method": "call_tool",
  "params": {
    "name": "query_mssql",
    "arguments": {
      "server": "sql.example.local",
      "database": "demo",
      "username": "readonly",
      "password": "***",
      "sql": "SELECT TOP (10) * FROM Sales;"
    }
  }
}

The server must reply with a JSON line that includes a result.content[] entry containing {"type": "json", "data": [...]} – exactly what the bundled Python code produces.

MCP Server Kurulumu

Refer to the following reference files if you need to deploy the MCP server yourself (both stdin/stdout and TCP JSON-line transports are supported).

server.py

import json
import asyncio
import pyodbc
from decimal import Decimal
from datetime import datetime, date, time

TCP_PORT = 5555   # n8n / Node buraya bağlanacak

# -------------------------------------------------------
# MSSQL Dinamik Bağlantı
# -------------------------------------------------------
def run_query_dynamic(server, database, username, password, sql):
    conn_str = (
        "DRIVER={ODBC Driver 17 for SQL Server};"
        f"SERVER={server};"
        f"DATABASE={database};"
        f"UID={username};"
        f"PWD={password};"
    )

    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute(sql)

    columns = [col[0] for col in cursor.description]
    results = []

    for row in cursor.fetchall():
        row_dict = {}
        for col, val in zip(columns, row):
            if isinstance(val, Decimal):
                row_dict[col] = float(val)  # Decimal -> float
            elif isinstance(val, (datetime, date, time)):
                row_dict[col] = val.isoformat()  # datetime/date/time -> ISO string
            else:
                row_dict[col] = val
        results.append(row_dict)

    conn.close()
    return results

# -------------------------------------------------------
# JSON-Line protokolünde gelen isteği işleme
# -------------------------------------------------------
async def process_request(line):
    print(f"\n--- GELEN İSTEK ---\n{line}\n-------------------", flush=True)

    try:
        request = json.loads(line)
        method = request.get("method")
        req_id = request.get("id")

        if method == "call_tool":
            tool_name = request["params"]["name"]
            args = request["params"]["arguments"]

            if tool_name == "query_mssql":
                server = args["server"]
                database = args["database"]
                username = args["username"]
                password = args["password"]
                sql = args["sql"]

                result = run_query_dynamic(server, database, username, password, sql)

                response = {
                    "id": req_id,
                    "result": {
                        "content": [
                            {
                                "type": "json",
                                "data": result
                            }
                        ]
                    }
                }

                return json.dumps(response)

    except Exception as e:
        return json.dumps({"error": str(e)})

    return None

# -------------------------------------------------------
# STDIN / STDOUT MCP Server
# -------------------------------------------------------
async def stdin_loop():
    loop = asyncio.get_event_loop()
    print("MCP server (stdin/stdout) ready...", flush=True)

    while True:
        line = await loop.run_in_executor(None, input)
        response = await process_request(line)
        if response:
            print(response, flush=True)

# -------------------------------------------------------
# TCP Server (JSON-Line)
# -------------------------------------------------------
async def handle_tcp(reader, writer):
    addr = writer.get_extra_info("peername")
    print(f"TCP bağlantısı geldi: {addr}")

    while True:
        try:
            data = await reader.readline()
            if not data:
                break

            line = data.decode().strip()
            if line == "":
                continue

            response = await process_request(line)
            if response:
                writer.write((response + "\n").encode())
                await writer.drain()

        except Exception as e:
            err = json.dumps({"error": str(e)})
            writer.write((err + "\n").encode())
            await writer.drain()

    print(f"TCP bağlantısı kapandı: {addr}")
    writer.close()
    await writer.wait_closed()

async def tcp_server():
    server = await asyncio.start_server(handle_tcp, "0.0.0.0", TCP_PORT)
    print(f"MCP TCP server started on port {TCP_PORT}")

    async with server:
        await server.serve_forever()

# -------------------------------------------------------
# MAIN
# -------------------------------------------------------
async def main():
    await asyncio.gather(
        stdin_loop(),
        tcp_server(),
    )

if __name__ == "__main__":
    asyncio.run(main())

mcp.json

{
  "name": "mssql_mcp_server",
  "version": "1.0.1",
  "tools": [
    {
      "name": "query_mssql",
      "description": "Dinamik MSSQL bağlantısı ile SQL sorgusu çalıştırır",
      "input_schema": {
        "type": "object",
        "properties": {
          "server": { "type": "string" },
          "database": { "type": "string" },
          "username": { "type": "string" },
          "password": { "type": "string" },
          "sql": { "type": "string" }
        },
        "required": ["server", "database", "username", "password", "sql"]
      }
    }
  ]
}

Packaging Tips

• Keep the server script and any virtual environments outside of dist; this package already ships with server/ and mcp.json.
• When deploying to n8n Cloud or Docker, ensure the Python binary and ODBC driver are installed inside the container.

License

MIT © Abdurrahim Ceylan