Search

Connectors

Contact support

Build a “Classic” MCP Server on Peliqan

You can write your own MCP (Model Context Protocol) Server, which communicates with the Peliqan Data Cloud. Your MCP Server can e.g. query data in the data warehouse, make real-time API calls and run automations.

You typically implement “tools” in a Python script, decorated with @mcp.tools(). We’ll walk you through the different steps below.

image
☝

Alternatively you can also use Peliqan’s Remote MCP, which allows you to write and maintain your MCP functions (tools) in Peliqan, in the cloud.

1. Set up your local MCP server project

Follow the below steps to set up an MCP server on your local computer.

Install uv:

curl -LsSf https://astral.sh/uv/install.sh | sh

Create a Python project:

# Create a new directory for the project
uv init peliqan-mcp
cd peliqan-mcp

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" httpx

# Create our server file
touch peliqan.py
☝

If you prefer not to use uv and no virtual environment, you can also simply use: pip install mcp pip install httpx

2. Code an MCP Server

Example code for the MCP server in peliqan.py:

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("peliqan")

API_BASE = "https://app.eu.peliqan.io"
API_TOKEN = "<your Peliqan API token>"

async def make_peliqan_request(url: str) -> dict[str, Any] | None:
    """Make a request to the Peliqan API"""
    headers = {
        "User-Agent": "peliqan-mcp",
        "Accept": "application/json",
        "Authorization": f"JWT {API_TOKEN}"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

@mcp.tool()
async def get_connections() -> str:
    """Get Connections.
    """
    url = f"{API_BASE}/api/servers/"
    data = await make_peliqan_request(url)

    if not data:
        return "Unable to fetch connections."
    
    response = []
    for item in data:
        response.append(item["name"])

    return "\n".join(response)

@mcp.tool()
async def read_data(tablename: str) -> str:
    """Read data from a table, via a custom API endpoint, defined in Peliqan.
    """
    url = f"https://api.eu.peliqan.io/<your_account_id>/myapi?tablename={tablename}"
    data = await make_peliqan_request(url)

    if not data:
        return "Unable to fetch data."
    
    return str(data)

if __name__ == "__main__":
    mcp.run(transport='stdio')

3. Create a custom API endpoint in Peliqan

The above function read_data calls a custom API endpoint, defined in your Peliqan account.

Example code for the custom API endpoint:

from urllib.parse import parse_qs

def handler(request):
    query_string = request['query_string']
    query_string_dict = parse_qs(query_string)
    if "tablename" in query_string_dict:
        tablename = query_string_dict['tablename'][0]
      
    dbconn = pq.dbconnect(pq.DW_NAME)
    query = f"SELECT id, name FROM odoo.{tablename} LIMIT 5"
    rows = dbconn.fetch(pq.DW_NAME, query = query)
    return {'rows': rows}

Usage of the custom API endpoint:

GET https://api.eu.peliqan.io/<your_peliqan_account_id>/myapi?tablename=partners
Authorization: JWT <your Peliqan API token>