> ## Documentation Index > Fetch the complete documentation index at: https://help.peliqan.io/llms.txt > Use this file to discover all available pages before exploring further. # Build a custom MCP Server on Peliqan > Build a custom MCP Server in Peliqan by adding Python tools for Text-to-SQL, RAG, writeback, intent detection, and user-level permissions. Peliqan has an [out-of-the-box MCP Server](/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()`. ## How to build a Custom MCP Server in Peliqan 1. Install the "MCP Server" script from the app templates: ![MCP Server template tile in Peliqan](https://mintcdn.com/peliqan-9d7f0393/6COfYI9KJIgEM3XW/images/Data%20Apps%20-%20low%20code%20Python/Build%20a%20custom%20MCP%20Server%20on%20Peliqan/template.png?fit=max\&auto=format\&n=6COfYI9KJIgEM3XW\&q=85\&s=18751b0e3cf490ddc72741b4b2130325) 2. 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](/publish-apis) on how to publish an API endpoint in Peliqan. ![API endpoint linked to MCP Server script](https://mintcdn.com/peliqan-9d7f0393/6COfYI9KJIgEM3XW/images/Data%20Apps%20-%20low%20code%20Python/Build%20a%20custom%20MCP%20Server%20on%20Peliqan/api_endpoint.png?fit=max\&auto=format\&n=6COfYI9KJIgEM3XW\&q=85\&s=6fcdd35395fd914c60cf85aead65f5fd) 3. 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: > > ![ChatGPT app creation with MCP endpoint URL](https://mintcdn.com/peliqan-9d7f0393/6COfYI9KJIgEM3XW/images/Data%20Apps%20-%20low%20code%20Python/Build%20a%20custom%20MCP%20Server%20on%20Peliqan/Screenshot_2025-12-30_at_15.57.28.png?fit=max\&auto=format\&n=6COfYI9KJIgEM3XW\&q=85\&s=a85889b308908a245c54bf70fcc1a41a) > > 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: > > ![Claude add custom connector with API key](https://mintcdn.com/peliqan-9d7f0393/6COfYI9KJIgEM3XW/images/Data%20Apps%20-%20low%20code%20Python/Build%20a%20custom%20MCP%20Server%20on%20Peliqan/Screenshot_2025-12-30_at_15.58.15.png?fit=max\&auto=format\&n=6COfYI9KJIgEM3XW\&q=85\&s=3c551d882905c1cd90dbe4e5afafc2c7) > **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: > > Add a security group (to which the user belongs) in Power Automate Admin center, for the setting Manage > Tenant settings > "Copilot studio authors". > > ![Power Automate Admin center tenant settings](https://mintcdn.com/peliqan-9d7f0393/o6KEmKVFeGGVHnKE/images/Peliqan%20MCP%20Server/Screenshot_2026-06-12_at_17.35.13.png?fit=max\&auto=format\&n=o6KEmKVFeGGVHnKE\&q=85\&s=9d0df77eeeee74c4347c6f532ca176e6) > > ![Power Automate Admin center security group selection](https://mintcdn.com/peliqan-9d7f0393/o6KEmKVFeGGVHnKE/images/Peliqan%20MCP%20Server/Screenshot_2026-06-12_at_17.35.19.png?fit=max\&auto=format\&n=o6KEmKVFeGGVHnKE\&q=85\&s=fb7f9b97e209810462c20874b48b1978) > > If you get an error "A custom connector with display name xxx already exists", follow the below steps: > > 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** > > ```python theme={null} > @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 > ``` ### RAG search More info: [Implement RAG](/build-ai-agents-in-peliqan/rag) ### Writeback 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** > > ```python theme={null} > @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}" > ``` ### 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** > > ```python theme={null} > # 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("'>","" > ``` ## 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, Peliqan OAuth) Implement SSO (single sign on) using OAuth in your MCP Server. More info: [Custom MCP with OAuth](/custom-mcp-with-o-auth) ### Authorization 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: ```python theme={null} # 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) ``` More info on the Peliqan permission model with groups: [Permissions](/permissions) #### Row-level data access See also: [Row-level data access](/permission-layer-for-ai-agents) > **Example MCP server with row-level access** > > ```python theme={null} > # 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 > ``` #### Permissions on MCP tool access 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** > > ```python theme={null} > # 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 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 > > 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) > 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 > ``` #### 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 This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.