Peliqan has an out-of-the-box MCP Server. However, you can also build custom MCP Servers in Peliqan. You can use the Peliqan MCP Server template to get started. You can extend this template by adding your own “tools”, these are Python functions decorated with @mcp.tool().
Install the “MCP Server” script from the app templates:
Add an API endpoint (method POST) and link it to the “MCP Server” script (this is the API handler script). Click here for more info on how to publish an API endpoint in Peliqan.
Update the credentials (API key) in your script and store in the Peliqan Secret Store.
Configure your MCP Client
‣
ChatGPT
In ChatGPT, click on your account name (bottom left corner) and click on Settings.
Go to Settings > Apps > Advanced Settings > Enable “Developer mode”.
Go to Settings > Apps > Create App.
Enter the URL of your MCP API endpoint, including the API key:
If you make updates to your MCP Server (e.g. you add tools), make sure to click the “Refresh” button on the detail screen of your app and check if the new actions are available for your app.
‣
Claude
You can use Claude.ai (in the browser) or Claude Desktop, but you need a Paid plan (e.g. Claude Pro) to add a Remote MCP Server.
Go to Settings > Connectors > Add custom connector.
Enter the URL of your MCP API endpoint, including the API key:
‣
Microsoft Copilot
In Copilot Studio, add an Agent and add a Tool of type “MCP Server” to the agent.
Select None or API key for the MCP Server authentication type (depending on how you implemented authorization in your custom MCP Server).
Publish your agent in Copilot Studio to make it available in Copilot chat. Next, you can access your Agent in Copilot chat and use it to access your MCP Server.
Troubleshooting in Copilot Studio
If you don’t have permissions to publish you agent, follow the below steps to allow a user to publish an agent:
‣
Click to expand
Add a security group (to which the user belongs) in Power Automate Admin center, for the setting Manage > Tenant settings > “Copilot studio authors”.
If you get an error “A custom connector with display name xxx already exists”, follow the below steps:
‣
Click to expand
Go to Power Automate > More > Discover all > Custom connectors. This will show a list of all “Tools” of type MCP Server that were added to your Copilot agents in the past. Delete old connectors here.
Implement capabilities
The Peliqan MCP Server template contains various examples on how to implement capabilities into your custom MCP Server such as Text-to-SQL data search for conversational analytics, RAG search on unstructured data, Writeback into business applications etc.
Text-to-SQL search
See full example in the MCP Server template.
‣
Example code (click to expand)
@mcp.tool()
def execute_query(query: str) -> List[Dict[str, Any]]:
"""
Executes an SQL query on the data warehouse and returns a list of rows.
"""
rows = dbconn.fetch(pq.DW_NAME, query = query)
return rows
Writeback is the concept of doing live API calls to a connected business application. This can be used to fetch live data or to take actions in a business application, e.g. add a deal in a CRM, create a draft invoice in accounting software etc.
‣
Example code (click to expand)
Other
See the MCP Server template for an example on how to implement your own custom tools. Implement them as Python functions and decorate with @mcp.tool().
Implement intent detection
Intent detection (logging the original intent of the user) is useful to improve your MCP implementation. See example code below. Without intent detection you will only be able to log tool invokations, without knowing what the user (using the MCP) wanted to achieve.
‣
Example custom MCP Server with intent detection (click to expand code)
Implement permissions
If you want to enforce permissions per user in your MCP Server, you have to authenticate individual users and apply authorization (permissions).
Authentication
You can authenticate users in your MCP Server using personal API keys or SSO using oAuth.
Personal API keys
Personal API keys
SSO using oAuth (Azure Entra, Google…)
Implement SSO (single sign on) using oAuth in your MCP Server:
You can apply permissions on various resources in your MCP Server code.
Permissions on data access
Use Peliqan groups, in order to limit access to datasets in your MCP Server. Put data resources (schemas and tables) in specific groups, map the authenticated user or group to a “Service account user” in Peliqan and use impersonation in your MCP Server.
Example impersonation in your MCP Server code:
More info on the Peliqan permission model with groups:
Limit access to certain MCP tools based on the authenticated user or a group of users. Below is an example using access levels to MCP tools.
‣
Example on how to add permissions on MCP tool usage (click to expand)
Permissions on Writeback
In order to implement permissions on Writeback (live API calls from the MCP Server to connected business applications), use private connections in Peliqan:
Create a group in Peliqan per user
Invite every user to add their own private connections in Peliqan (e.g. to your CRM). Use the “Send invite” feature when adding a connection.
Add these private connections to the specific groups (one group per user)
Expose the private connections based on the authenticated user in your MCP Server
@mcp.tool()
def pipedrive_add_contact(*args, **kwargs):
"""
Example of a writeback function: add a contact to Pipedrive (CRM).
"""
pipedrive_api = pq.connect('Pipedrive')
person = {}
if "name" in kwargs:
person["name"] = kwargs["name"]
if "email" in kwargs:
person["emails"] = [{'label': 'work', 'value': kwargs["email"], 'primary': True}]
result = pipedrive_api.add('person', person)
if "status" in result and result["status"] == "success":
new_contact = result.get("detail", {}).get("data", {})
return f"Contact added to Pipedrive: {new_contact}"
else:
print("Response from Pipedrive:", result)
error = result.get("detail", {}).get("response_json", {}).get("error", "")
return f"Error adding contact to Pipedrive: {error}"
# Add an API endpoint with method POST and link it to this API handler script.
# Use your MCP Server in e.g. ChatGPT or Claude: add a Remote MCP server with the URL API endpoint set for this script.
# Example: POST https://api.eu.peliqan.io/123/mcp?api_key=xxxx
#
# Add custom MCP tools: see below section, decorate your function with @mcp.tool()
# Version 2.2
# Change below API key and put it in Peliqan's secret store (under Connections)
# api_key = pq.get_secret("MCP API key")
api_key = "xxxxx" # for demo only, remove this
import json
import inspect
import time
from typing import get_type_hints
from typing import List, Dict, Any
from urllib.parse import parse_qs
dbconn = pq.dbconnect(pq.DW_NAME)
TOOL_INTENT_DESCRIPTION = (
"Explain the original intent of the user, taking into account previous instructions from the user,"
"so that the full intent from the user and reason to invoke this tool is clear. Never include credentials"
)
MCP_TOOLS = []
class mcp:
def tool():
def decorator(func):
sig = inspect.signature(func)
type_hints = get_type_hints(func)
# parse descriptions from docstring using :param style
raw_doc = func.__doc__ or ""
param_docs = {}
for line in raw_doc.splitlines():
line = line.strip()
if line.startswith(":param"):
# example: ":param first_name: the user's first name"
try:
_, rest = line.split("param", 1)
name, desc = rest.split(":", 1)
param_docs[name.strip()] = desc.strip()
except ValueError:
pass
properties = {}
required = []
for name, param in sig.parameters.items():
hint = type_hints.get(name, "any")
type_str = hint.__name__ if isinstance(hint, type) else str(hint)
if type_str == "str":
mcp_type_str = "string"
elif type_str == "int":
mcp_type_str = "number"
else:
mcp_type_str = "string"
prop = {
"type": mcp_type_str,
"description": param_docs.get(name, "")
}
if param.default is not inspect._empty:
prop["default"] = param.default
else:
required.append(name)
properties[name] = prop
# Intent detection: injected into every tool's schema here, rather
# than added as a real parameter to each function individually.
# Stripped back out before calling the real function.
properties["tool_intent"] = {
"type": "string",
"description": TOOL_INTENT_DESCRIPTION,
}
MCP_TOOLS.append({
"name": func.__name__,
"description": raw_doc.strip().split("\n")[0] if raw_doc else "",
"inputSchema": {
"type": "object",
"properties": properties,
"required": required
}
})
return func
return decorator
def _log_tool_intent(tool_name, tool_intent):
"""
Log the calling agent's stated intent for this tool call.
"""
if not tool_intent:
print("no tool_intent")
return
try:
dbconn.write("logs", "tool_intent_log", [{
"logged_at": str(int(time.time())),
"tool_name": tool_name,
"tool_intent": tool_intent,
}], pk="logged_at", object_schema={
"properties": {
"logged_at": {"type": "string"},
"tool_name": {"type": "string"},
"tool_intent": {"type": "string"},
}
})
except Exception:
pass
def get_tool_response_format(tool_name):
func = globals().get(tool_name)
response_annotation = inspect.signature(func).return_annotation
if hasattr(response_annotation, "__name__"):
return response_annotation.__name__
return str(response_annotation).replace("typing.", "").replace("class '", "").replace("'>",""
########################### ADD MCP TOOLS BELOW ###########################
@mcp.tool()
def say_hello(first_name: str, last_name: str = "") -> str:
"""
Peliqan will say hello.
:param first_name: the user's first name
:param last_name: optional last name
"""
return f"Hi there {first_name} {last_name} from Peliqan MCP server !"
@mcp.tool()
def execute_query(query: str) -> List[Dict[str, Any]]:
"""
Executes an SQL query on the data warehouse and returns a list of rows.
"""
query = query.replace('"' + pq.DW_NAME + '".', '').replace(pq.DW_NAME + '.', '')
print("Final SQL query to execute:")
print(query)
rows = dbconn.fetch(pq.DW_NAME, query = query)
return rows
########################### END OF MCP TOOLS ###########################
def log_request(request):
print("request method: ", request['method'])
print("request url: ", request['url'])
print("request query string: ", request['query_string'])
print("request body:")
try:
print(json.dumps(request['data'], indent=2))
except:
print(request['data'])
def log_response(response):
print("Response:")
print(json.dumps(response, indent=2))
def mcp_response_initialize(id):
response_initialize = {
"jsonrpc": "2.0",
"id": id,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"callTool": True,
"listTools": True,
"tools": {
"listChanged": False
}
},
"serverInfo": {
"name": "peliqan-mcp",
"version": "0.0.1"
}
}
}
return response_initialize
def mcp_response_tools_list(id):
response_tools_list = {
"jsonrpc": "2.0",
"id": id,
"result": {
"tools": MCP_TOOLS
}
}
return response_tools_list
def mcp_response_tools_call(id, response_type):
response_tools_call = {
"jsonrpc": "2.0",
"id": id,
"result": {
"content": [
{
"type": response_type,
response_type: ""
}
],
"isError": False
}
}
return response_tools_call
def check_api_key(request):
query_string_parsed = parse_qs(request.get("query_string", ""))
request_api_key = query_string_parsed["api_key"][0] if "api_key" in query_string_parsed else None
if request_api_key != api_key:
print("Incorrect API key, stopping")
return False
else:
return True
def handler(request):
log_request(request)
if not check_api_key(request):
return "Unauthorized", 401 #, {"WWW-Authenticate": 'Bearer resource_metadata="https://your-server.com/"'}
data = request['data']
if not data:
data = "{}"
mcp_req = json.loads(data)
id = 0
if "id" in mcp_req:
id = mcp_req["id"]
mcp_response = mcp_response_initialize(id)
if "method" in mcp_req:
if mcp_req["method"] == "initialize":
response = mcp_response_initialize(id)
elif mcp_req["method"] == "notifications/initialized":
return "", 202
elif mcp_req["method"] == "tools/list":
mcp_response = mcp_response_tools_list(id)
elif mcp_req["method"] == "tools/call":
tool_name = mcp_req["params"]["name"]
args = dict(mcp_req["params"].get("arguments", {}))
tool_intent = args.pop("tool_intent", "") # Remove for actual function call
_log_tool_intent(tool_name, tool_intent)
isError = False
try:
tool_response = globals()[tool_name](**args) #Invoking tool
except Exception as e:
print(e)
tool_response = str(e)
isError = True
tool_response_format = get_tool_response_format(tool_name) # str, List
response_type = "text"
mcp_response = mcp_response_tools_call(id, response_type)
if isError:
mcp_response["result"]["isError"] = isError
if tool_response_format == "str":
mcp_response["result"]["content"][0][response_type] = tool_response
else: # List
mcp_response["result"]["content"][0][response_type] = json.dumps(tool_response)
log_response(mcp_response)
return mcp_response
# CREATE SERVICE ACCOUNT USERS IN PELIQAN
# Create users in Peliqan and assign them to a group.
# Create an API key for each user and add to a Secret Store in Peliqan.
# IMPLEMENT AUTHENTICATION
# 1. Implement authentication in your MCP Server,
# e.g. using SSO with Microsoft Azure Entra.
# 2. Map the authenticated user or group to Service account users in Peliqan,
# use for example read the group claims from Azure Entra.
user_group_mappings = {
"usergroup1": pq.get_secret("ServiceAccountApiKey1"),
"usergroup2": pq.get_secret("ServiceAccountApiKey2"),
}
# Apply user impersonation
current_user_group = "xxx" # read e.g. groups claim from SSO with Azure Entra
peliqan_api_key = user_group_mappings[current_user_group]
personal_pq = Peliqan(peliqan_api_key) # Create a pq instance with impersonation
# Use personal_pq to fetch data from the data warehouse
personal_dbconn = personal_pq.dbconnect(pq.DW_NAME)
query = "SELECT * FROM some_table" # e.g. from text-to-SQL search
rows = personal_dbconn.fetch(personal_pq.DW_NAME, query = query)
# Change below API keys and put them in Peliqan's secret store (under Connections) for production use.
api_key_1 = "df10c215-274e-4672-938a-7d264beac434" # Bob - for demo only, move to secret store
api_key_2 = "1a9d5eb4-93f7-4de3-b6aa-67ebf22e7bc0" # Lucas - for demo only, move to secret store
API_KEYS = {
api_key_1: "bob",
api_key_2: "lucas",
}
import json
import inspect
from typing import get_type_hints
from typing import List, Dict, Any
from urllib.parse import parse_qs
CURRENT_USERNAME = None
dbconn = pq.dbconnect(pq.DW_NAME)
# Set the schema that contains "access views" in Peliqan,
# these are views (queries) that select source data, join it with a
# permission table and that contain a placeholder {username}
# that will be replaced here in the MCP Server in tool execute_query
MCP_ACCESS_SCHEMA = "MCP Access"
MCP_TOOLS = []
class mcp:
def tool():
def decorator(func):
sig = inspect.signature(func)
type_hints = get_type_hints(func)
# parse descriptions from docstring using :param style
raw_doc = func.__doc__ or ""
param_docs = {}
for line in raw_doc.splitlines():
line = line.strip()
if line.startswith(":param"):
try:
_, rest = line.split("param", 1)
name, desc = rest.split(":", 1)
param_docs[name.strip()] = desc.strip()
except ValueError:
pass
properties = {}
required = []
for name, param in sig.parameters.items():
hint = type_hints.get(name, "any")
type_str = hint.__name__ if isinstance(hint, type) else str(hint)
if type_str == "str":
mcp_type_str = "string"
elif type_str == "int":
mcp_type_str = "number"
else:
mcp_type_str = "string"
prop = {
"type": mcp_type_str,
"description": param_docs.get(name, "")
}
if param.default is not inspect._empty:
prop["default"] = param.default
else:
required.append(name)
properties[name] = prop
MCP_TOOLS.append({
"name": func.__name__,
"description": raw_doc.strip().split("\n")[0] if raw_doc else "",
"inputSchema": {
"type": "object",
"properties": properties,
"required": required
}
})
return func
return decorator
def get_tool_response_format(tool_name):
func = globals().get(tool_name)
response_annotation = inspect.signature(func).return_annotation
if hasattr(response_annotation, "__name__"):
return response_annotation.__name__
return str(response_annotation).replace("typing.", "").replace("class '", "").replace("'>","")
########################### ADD MCP TOOLS BELOW ###########################
@mcp.tool()
def list_tables() -> List[Dict[str, Any]]:
"""
Returns the access views available through this MCP server, discovered live
from the "MCP Access" schema. Each one is already scoped to the authenticated
caller's row-level access permissions - there is no way to query the underlying
source tables directly.
"""
views = []
for db in pq.list_databases():
for table in db["tables"]:
schema_name = next((s["name"] for s in db["schemas"] if s["id"] == table["schema_id"]), None)
if schema_name != MCP_ACCESS_SCHEMA:
continue
table_meta = pq.get_table(table["id"])
views.append({"table": table["name"], "query": table_meta.get("query", "") or ""})
return views
@mcp.tool()
def execute_query(query: str) -> List[Dict[str, Any]]:
"""
Executes an SQL query on the data warehouse and returns a list of rows.
Row-level access views (see list_tables) are automatically prepended as CTEs,
scoped to the authenticated caller - only SELECT is allowed.
"""
query = query.replace('"' + pq.DW_NAME + '".', '').replace(pq.DW_NAME + '.', '')
cte_defs = ",\n".join(
f"{view['table']} AS ({view['query'].replace('{username}', CURRENT_USERNAME)})"
for view in list_tables()
)
query = f"WITH {cte_defs}\n{query}"
print("Final SQL query to execute:")
print(query)
rows = dbconn.fetch(pq.DW_NAME, query = query)
return rows
########################### END OF MCP TOOLS ###########################
def log_request(request):
print("request method: ", request['method'])
print("request url: ", request['url'])
print("request query string: ", request['query_string'])
print("request body:")
try:
print(json.dumps(request['data'], indent=2))
except:
print(request['data'])
def log_response(response):
print("Response:")
print(json.dumps(response, indent=2))
def mcp_response_initialize(id):
return {
"jsonrpc": "2.0",
"id": id,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"callTool": True,
"listTools": True,
"tools": {"listChanged": False}
},
"serverInfo": {"name": "peliqan-mcp", "version": "0.0.1"}
}
}
def mcp_response_tools_list(id):
return {"jsonrpc": "2.0", "id": id, "result": {"tools": MCP_TOOLS}}
def mcp_response_tools_call(id, response_type):
return {
"jsonrpc": "2.0",
"id": id,
"result": {
"content": [{"type": response_type, response_type: ""}],
"isError": False
}
}
def check_api_key(request):
global CURRENT_USERNAME
query_string_parsed = parse_qs(request.get("query_string", ""))
request_api_key = query_string_parsed["api_key"][0] if "api_key" in query_string_parsed else None
username = API_KEYS.get(request_api_key)
if not username:
print("Incorrect or unknown API key, stopping")
return False
CURRENT_USERNAME = username
return True
def handler(request):
log_request(request)
if not check_api_key(request):
return "Unauthorized", 401
data = request['data']
if not data:
data = "{}"
mcp_req = json.loads(data)
id = mcp_req.get("id", 0)
mcp_response = mcp_response_initialize(id)
method = mcp_req.get("method")
if method == "initialize":
mcp_response = mcp_response_initialize(id)
elif method == "notifications/initialized":
return "", 202
elif method == "tools/list":
mcp_response = mcp_response_tools_list(id)
elif method == "tools/call":
tool_name = mcp_req["params"]["name"]
args = mcp_req["params"].get("arguments", {})
isError = False
try:
tool_response = globals()[tool_name](**args)
except Exception as e:
print(e)
tool_response = str(e)
isError = True
tool_response_format = get_tool_response_format(tool_name)
response_type = "text"
mcp_response = mcp_response_tools_call(id, response_type)
if isError:
mcp_response["result"]["isError"] = isError
if tool_response_format == "str":
mcp_response["result"]["content"][0][response_type] = tool_response
else:
mcp_response["result"]["content"][0][response_type] = json.dumps(tool_response)
log_response(mcp_response)
return mcp_response
# Add an API endpoint with method POST and link it to this API handler script.
# Use your MCP Server in e.g. ChatGPT or Claude: add a Remote MCP server with the URL API endpoint set for this script.
# Example: POST https://api.eu.peliqan.io/123/mcp?api_key=82ac4ac4-8ab6-4c2c-a039-548d19387d1d
#
# Add custom MCP tools: see below section, decorate your function with @mcp.tool()
# Version 2.2
# Change below API key and put it in Peliqan's secret store (under Connections)
# api_key = pq.get_secret("MCP API key access level x")
# for demo only, remove the keys below
api_key_hello = "82ac4ac4-8ab6-4c2c-a039-548d19387d1d" # say_hello tool
api_key_list = "8117afec-6318-431a-b8b8-11f8f7cb3572" # say_hello + all list tools
api_key_all = "bc438e14-09dd-4750-85cf-93ab4119a1d9" # all tools
import json
import inspect
from typing import get_type_hints
from typing import List, Dict, Any
from urllib.parse import parse_qs
dbconn = pq.dbconnect(pq.DW_NAME)
API_KEY_ACCESS_LEVELS = {
api_key_hello: 1,
api_key_list: 2,
api_key_all: 3,
}
MCP_TOOLS = []
TOOL_PERMISSIONS = {}
class mcp:
def tool(min_access_level: int=1):
def decorator(func):
sig = inspect.signature(func)
type_hints = get_type_hints(func)
# parse descriptions from docstring using :param style
raw_doc = func.__doc__ or ""
param_docs = {}
for line in raw_doc.splitlines():
line = line.strip()
if line.startswith(":param"):
# example: ":param first_name: the user's first name"
try:
_, rest = line.split("param", 1)
name, desc = rest.split(":", 1)
param_docs[name.strip()] = desc.strip()
except ValueError:
pass
properties = {}
required = []
for name, param in sig.parameters.items():
hint = type_hints.get(name, "any")
type_str = hint.__name__ if isinstance(hint, type) else str(hint)
if type_str == "str":
mcp_type_str = "string"
elif type_str == "int":
mcp_type_str = "number"
else:
mcp_type_str = "string"
prop = {
"type": mcp_type_str,
"description": param_docs.get(name, "")
}
if param.default is not inspect._empty:
prop["default"] = param.default
else:
required.append(name)
properties[name] = prop
MCP_TOOLS.append({
"name": func.__name__,
"description": raw_doc.strip().split("\n")[0] if raw_doc else "",
"inputSchema": {
"type": "object",
"properties": properties,
"required": required
}
})
TOOL_PERMISSIONS[func.__name__] = min_access_level
return func
return decorator
def get_tool_response_format(tool_name):
func = globals().get(tool_name)
response_annotation = inspect.signature(func).return_annotation
if hasattr(response_annotation, "__name__"):
return response_annotation.__name__
return str(response_annotation).replace("typing.", "").replace("class '", "").replace("'>","")
########################### ADD MCP TOOLS BELOW ###########################
@mcp.tool()
def say_hello(first_name: str, last_name: str = "") -> str:
"""
Peliqan will say hello.
:param first_name: the user's first name
:param last_name: optional last name
"""
return f"Hi there {first_name} {last_name} from Peliqan MCP server !"
@mcp.tool(min_access_level=2)
def list_connections() -> List[str]:
"""
Returns list of all ELT connections in the Peliqan account.
"""
connections = pq.list_connections()
conn_names = []
for connection in connections:
conn_names.append(connection["name"])
#return ", ".join(conn_names)
return conn_names
@mcp.tool(min_access_level=2)
def list_tables() -> List[Dict[str, Any]]:
"""
Returns list of all the tables in the Peliqan account.
"""
all_tables = []
for db in pq.list_databases():
for table in db["tables"]:
schema_name = next((s["name"] for s in db["schemas"] if s["id"] == table["schema_id"]), None)
all_tables.append({
"db_id": db["id"],
"db": db["name"],
"schema_id": table["schema_id"],
"schema": schema_name,
"table_id": table["id"],
"table": table["name"]
})
return all_tables
@mcp.tool(min_access_level=2)
def list_columns(table_id: int) -> List[str]:
"""
Returns list of all the fields (columns) of a table.
"""
table_meta = pq.get_table(table_id)
columns = table_meta.get("all_fields", [])
column_names = [col["name"] for col in columns if not col["name"].startswith("_sdc")]
return column_names
@mcp.tool(min_access_level=3)
def execute_query(query: str) -> List[Dict[str, Any]]:
"""
Executes an SQL query on the data warehouse and returns a list of rows.
"""
query = query.replace('"' + pq.DW_NAME + '".', '').replace(pq.DW_NAME + '.', '')
print("Final SQL query to execute:")
print(query)
rows = dbconn.fetch(pq.DW_NAME, query = query)
return rows
@mcp.tool(min_access_level=3)
def pipedrive_add_contact(*args, **kwargs):
"""
Example of a writeback function: add a contact to Pipedrive (CRM).
"""
pipedrive_api = pq.connect('Pipedrive')
person = {}
if "name" in kwargs:
person["name"] = kwargs["name"]
if "email" in kwargs:
person["emails"] = [{'label': 'work', 'value': kwargs["email"], 'primary': True}]
result = pipedrive_api.add('person', person)
if "status" in result and result["status"] == "success":
new_contact = result.get("detail", {}).get("data", {})
return f"Contact added to Pipedrive: {new_contact}"
else:
print("Response from Pipedrive:", result)
error = result.get("detail", {}).get("response_json", {}).get("error", "")
return f"Error adding contact to Pipedrive: {error}"
########################### END OF MCP TOOLS ###########################
def log_request(request):
print("request method: ", request['method'])
print("request url: ", request['url'])
print("request query string: ", request['query_string'])
print("request body:")
try:
print(json.dumps(request['data'], indent=2))
except:
print(request['data'])
def log_response(response):
print("Response:")
print(json.dumps(response, indent=2))
def mcp_response_initialize(id):
response_initialize = {
"jsonrpc": "2.0",
"id": id,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"callTool": True,
"listTools": True,
"tools": {
"listChanged": False
}
},
"serverInfo": {
"name": "peliqan-mcp",
"version": "0.0.1"
}
}
}
return response_initialize
def mcp_response_tools_list(id, access_level):
visible_tools = [t for t in MCP_TOOLS if TOOL_PERMISSIONS.get(t["name"], 1) <= access_level]
response_tools_list = {
"jsonrpc": "2.0",
"id": id,
"result": {
"tools": visible_tools
}
}
return response_tools_list
def mcp_response_tools_call(id, response_type):
response_tools_call = {
"jsonrpc": "2.0",
"id": id,
"result": {
"content": [
{
"type": response_type,
response_type: ""
}
],
"isError": False
}
}
return response_tools_call
def get_api_key_access_level(request):
query_string_parsed = parse_qs(request.get("query_string", ""))
request_api_key = query_string_parsed["api_key"][0] if "api_key" in query_string_parsed else None
access_level = API_KEY_ACCESS_LEVELS.get(request_api_key)
if access_level is None:
print("Incorrect API key, stopping")
return None
else:
return access_level
def handler(request):
log_request(request)
access_level = get_api_key_access_level(request)
if not access_level:
return "Unauthorized", 401 #, {"WWW-Authenticate": 'Bearer resource_metadata="https://your-server.com/"'}
data = request['data']
if not data:
data = "{}"
mcp_req = json.loads(data)
id = 0
if "id" in mcp_req:
id = mcp_req["id"]
mcp_response = mcp_response_initialize(id)
if "method" in mcp_req:
if mcp_req["method"] == "initialize":
response = mcp_response_initialize(id)
elif mcp_req["method"] == "notifications/initialized":
return "", 202
elif mcp_req["method"] == "tools/list":
mcp_response = mcp_response_tools_list(id, access_level)
elif mcp_req["method"] == "tools/call":
tool_name = mcp_req["params"]["name"]
required_access_level = TOOL_PERMISSIONS.get(tool_name, 1)
if required_access_level > access_level:
return "Unauthorized for this tool", 401
args = {}
if "arguments" in mcp_req["params"]:
args = mcp_req["params"]["arguments"]
isError = False
try:
tool_response = globals()[tool_name](**args) #Invoking tool
except Exception as e:
print(e)
tool_response = str(e)
isError = True
tool_response_format = get_tool_response_format(tool_name) # str, List
response_type = "text"
mcp_response = mcp_response_tools_call(id, response_type)
if isError:
mcp_response["result"]["isError"] = isError
if tool_response_format == "str":
mcp_response["result"]["content"][0][response_type] = tool_response
else: # List
mcp_response["result"]["content"][0][response_type] = json.dumps(tool_response)
log_response(mcp_response)
return mcp_response
