Overview
Prompted by a customer question, I wanted to see what was possible in terms of MCP integration into Genie Code, in order to try this out I decided to look at Azure Dev Ops, as it's a common workflow to want to see your tickets alongside the work you're doing. When I dug into it I realised the Azure out of the box MCP did not have the right protocols to enable Genie code to use it. Given the common nature of this kind of request I thought I'd share my learnings here. After some testing I found the best way to achieve this was using a python udf in devops and then setting this up as a Genie Code MCP server. If you want to give this a go, here are the instructions and code.
What You'll Build
- Two Python UDFs in Unity Catalog that connect to Azure DevOps โ search_work_items and list_project_backlog. No extra infrastructure. Add them to Genie Code via the UC Function browser and start asking natural language questions immediately.
Prerequisites
- A Databricks workspace with Apps enabled
- Databricks CLI installed and authenticated
- An Azure DevOps organisation with a project
- A Personal Access Token (PAT) for Azure DevOps with Work Items (Read) scope
Step 1: Create an Azure DevOps Project & Personal token
Create DevOps Project
- Go to https://dev.azure.com and sign in (or create a free account)
- Create a new organisation
- Create a project
- Add some work items (Epics, Issues, Tasks) to have data to query
Create a Personal Access Token
- Click the user settings icon (top right) โ Personal access tokens
- Click New Token
- Name: <set-your-name>
- Scopes: Full access (tighten for production)
- Copy the token โ you'll need it in Step 3
Step 2: Enable Prerequisites
Before creating the UC Functions, enable outbound networking for UDFs:
In your Databricks workspace, go to Settings > Preview features
Enable "Enable networking for UDFs in Serverless SQL Warehouses"
Restart your SQL warehouse after enabling. Changes can take up to 15 minutes to propagate.
Step 3: Create the UC Functions
First, create a schema to hold the functions:
CREATE SCHEMA IF NOT EXISTS <your-catalog>.<your-schema>
COMMENT 'Azure DevOps MCP tools for Genie Code';
Note: Python UDFs do not support default parameter values โ all parameters are required.
Function 1: search_work_items
CREATE OR REPLACE FUNCTION <your-catalog>.<your-schema>.search_work_items(
query STRING COMMENT 'Text to search for in work item titles and descriptions',
max_results INT COMMENT 'Maximum number of results to return, e.g. 20'
)
RETURNS STRING
LANGUAGE PYTHON
COMMENT 'Search Azure DevOps work items by text in title or description. Use this to find tasks, issues, and epics matching a keyword.'
AS $$
import urllib.request, json, base64
PAT = "<your-pat-token>"
ORG = "<your-org-name>"
PROJECT = "<your-project-name>"
BASE = f"https://dev.azure.com/{ORG}"
creds = base64.b64encode(f":{PAT}".encode()).decode()
def api_call(path, body=None, extra_params=''):
url = f"{BASE}/{path}?api-version=7.0{extra_params}"
headers = {"Authorization": f"Basic {creds}", "Content-Type": "application/json"}
if body:
req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers)
else:
req = urllib.request.Request(url, headers=headers)
return json.loads(urllib.request.urlopen(req, timeout=30).read())
wiql = f"SELECT [System.Id] FROM workitems WHERE [System.Title] CONTAINS '{query}' OR [System.Description] CONTAINS '{query}' ORDER BY [System.ChangedDate] DESC"
result = api_call(f"{PROJECT}/_apis/wit/wiql", {"query": wiql}, f"&$top={max_results}")
ids = [wi["id"] for wi in result.get("workItems", [])]
if not ids:
return f"No work items found matching '{query}'."
fields = "System.Id,System.Title,System.State,System.AssignedTo,System.WorkItemType"
id_str = ",".join(str(x) for x in ids[:200])
data = api_call(f"{PROJECT}/_apis/wit/workitems", extra_params=f"&ids={id_str}&$fields={fields}")
lines = [f"Search results for '{query}' ({len(data.get('value', []))} items):", ""]
for item in data.get("value", []):
f = item.get("fields", {})
assigned = f.get("System.AssignedTo", {})
if isinstance(assigned, dict):
assigned = assigned.get("displayName", "Unassigned")
lines.append(f"- #{item['id']} [{f.get('System.WorkItemType', '?')}] {f.get('System.Title', 'Untitled')} - {f.get('System.State', '?')} (Assigned: {assigned})")
return "\n".join(lines)
$$;
Function 2: list_project_backlog
CREATE OR REPLACE FUNCTION <your-catalog>.<your-schema>.list_project_backlog(
max_results INT COMMENT 'Maximum number of backlog items to return, e.g. 50'
)
RETURNS STRING
LANGUAGE PYTHON
COMMENT 'List the active project backlog showing all open Epics, Issues, and Tasks grouped by type.'
AS $$
import urllib.request, json, base64
PAT = "<your-pat-token>"
ORG = "<your-org-name>"
PROJECT = "<your-project-name>"
BASE = f"https://dev.azure.com/{ORG}"
creds = base64.b64encode(f":{PAT}".encode()).decode()
def api_call(path, body=None, extra_params=''):
url = f"{BASE}/{path}?api-version=7.0{extra_params}"
headers = {"Authorization": f"Basic {creds}", "Content-Type": "application/json"}
if body:
req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers)
else:
req = urllib.request.Request(url, headers=headers)
return json.loads(urllib.request.urlopen(req, timeout=30).read())
wiql = "SELECT [System.Id] FROM workitems WHERE [System.WorkItemType] IN ('Epic', 'Issue', 'Task') AND [System.State] NOT IN ('Closed', 'Removed', 'Done') ORDER BY [Microsoft.VSTS.Common.BacklogPriority] ASC"
result = api_call(f"{PROJECT}/_apis/wit/wiql", {"query": wiql}, f"&$top={max_results}")
ids = [wi["id"] for wi in result.get("workItems", [])]
if not ids:
return "The backlog is empty."
fields = "System.Id,System.Title,System.State,System.AssignedTo,System.WorkItemType"
id_str = ",".join(str(x) for x in ids[:200])
data = api_call(f"{PROJECT}/_apis/wit/workitems", extra_params=f"&ids={id_str}&$fields={fields}")
by_type = {}
for item in data.get("value", []):
wtype = item.get("fields", {}).get("System.WorkItemType", "Other")
by_type.setdefault(wtype, []).append(item)
type_order = ["Epic", "Issue", "Task"]
lines = [f"Active Backlog ({len(data.get('value', []))} items):", ""]
for wtype in type_order:
items = by_type.get(wtype, [])
if items:
lines.append(f"{wtype}s ({len(items)}):")
for item in items:
f = item.get("fields", {})
assigned = f.get("System.AssignedTo", {})
if isinstance(assigned, dict):
assigned = assigned.get("displayName", "Unassigned")
lines.append(f" - #{item['id']} {f.get('System.Title', 'Untitled')} - {f.get('System.State', '?')} (Assigned: {assigned})")
lines.append("")
return "\n".join(lines)
$$;
Step 4: Test the Functions
SELECT <your-catalog>.<your-schema>.search_work_items('inspection', 5);
SELECT <your-catalog>.<your-schema>.list_project_backlog(50);
Step 5: Add to Genie Code
- Open Genie Code in your workspace
- Switch to Agent Mode
- Click Settings (gear icon)
- Under "Unity Catalog Function", click Browse
- Navigate to mcga_catalog.devops_tools
- Select the functions and Save
Result
You can now ask Genie Code natural language questions like:
- "What's on the backlog?"
- "Find work items related to vessel inspections"
- "Search for anything about compliance"
Key Notes for UC Functions Approach
- Python UDFs cannot have default parameter values โ all params must be required
- The "Enable networking for UDFs" preview must be enabled for outbound HTTP calls
- Functions use urllib.request (standard library) โ no pip dependencies needed
- The Azure DevOps PAT is embedded in the function body. For production, consider using UC service credentials instead.