Build an MCP tool

Expose Xorq UDXFs as tools Claude can call through the Model Context Protocol

This tutorial shows you how to expose Xorq functions as tools that Claude can call conversationally. You’ll learn how to use the Model Context Protocol (MCP) to integrate Xorq with Claude Desktop.

After completing this tutorial, you know how to wrap Xorq UDXFs as MCP tools, configure Claude Desktop to use them, and interact with your data functions through natural language.

Prerequisites

You need:

• Xorq installed (see Install Xorq)
• FastMCP library: pip install fastmcp
• OpenAI Python client: pip install openai
• OpenAI API key (set as environment variable)
• Claude Desktop: Download from claude.ai/download

Set up your API key

Before writing code, set your OpenAI API key as an environment variable.

macOS / Linux

export OPENAI_API_KEY="your-api-key-here"

Windows

set OPENAI_API_KEY=your-api-key-here

Tip

Never commit API keys to version control. Use environment variables or a secrets manager in production.

How this tutorial works

You’ll build a Python file incrementally. Each section adds new code to the file.

Two tabs per code block:

• Complete code: The full runnable file at this stage
• Changes: Just the lines you’re adding (shown as diff)

Create a file called sentiment_mcp_server.py and build it section by section.

For Claude Desktop integration: After completing the tutorial, you’ll configure Claude Desktop to use your completed script.

Tip

Think of MCP as a universal adapter. Claude speaks MCP to request tool executions. Your Xorq functions speak Flight protocol. FlightMCPServer translates between them, exposing Xorq UDXFs as tools Claude can call.

How the integration works

The architecture connects four components:

• Claude Desktop asks questions and calls tools through MCP.

• MCP server (FlightMCPServer) receives tool calls and translates them to Flight protocol.

• Flight server executes UDXFs (User-Defined Exchange Functions) from Xorq.

• UDXF processes data and returns results.

The flow looks like this:

User → Claude Desktop → MCP protocol → FlightMCPServer → Flight protocol → UDXF → Results → Claude

Understanding this flow helps you debug integration issues. If Claude can’t call your tool, then check each connection point: MCP server running? Flight server started? UDXF registered?

Create the FlightMCPServer class

Now you’ll build the bridge between MCP and Flight protocols. This class handles the translation.

Add this to your sentiment_mcp_server.py file:

Complete code

This is your complete sentiment_mcp_server.py file at this stage. It defines the FlightMCPServer class that bridges MCP and Flight protocols, handles UDXF registration, and manages communication between Claude Desktop and your Xorq functions.

from typing import Callable, Optional 
import toolz
from mcp.server.fastmcp import FastMCP
from xorq.flight import FlightServer, FlightUrl


class FlightMCPServer:
    def __init__(
        self,
        name: str,
        flight_port: int = 8818,
    ):

        self.name = name
        self.port = flight_port
        self.mcp = FastMCP(name)
        

        self.flight_server = None
        self.client = None
        

        self.udxfs = {}
        self.schemas = {}
        self.exchange_functions = {}
    

    def start_flight_server(self) -> FlightServer:
        if self.flight_server:
            return self.flight_server
        
        try:

            self.flight_server = FlightServer(
                FlightUrl(port=self.port), 
                exchangers=list(self.udxfs.values())
            )
            

            self.flight_server.serve()
            

            self.client = self.flight_server.client
            

            for udxf_name, udxf in self.udxfs.items():
                self.exchange_functions[udxf_name] = toolz.curry(
                    self.client.do_exchange, udxf.command
                )
            
            return self.flight_server
        except Exception:
            raise
    

    def create_mcp_tool(
        self,
        udxf,
        input_mapper: Callable,
        tool_name: Optional[str] = None,
        description: Optional[str] = None,
        output_mapper: Optional[Callable] = None,
    ):

        udxf_command = udxf.command
        tool_name = tool_name or udxf_command
        

        self.udxfs[udxf_command] = udxf
        

        if not self.flight_server:
            self.start_flight_server()
        

        do_exchange = self.exchange_functions.get(udxf_command)
        

        if output_mapper is None:
            def default_output_mapper(result_df):
                if len(result_df) > 0:
                    return result_df.to_string()
                return "No results"
            
            actual_output_mapper = default_output_mapper
        else:
            actual_output_mapper = output_mapper
        

        @self.mcp.tool(name=tool_name, description=description)
        async def wrapper(**kwargs):
            try:

                input_data = input_mapper(**kwargs)
                

                _, result = do_exchange(input_data.to_pyarrow_batches())
                result_df = result.read_pandas()
                

                output = actual_output_mapper(result_df)
                return output
            except Exception as e:
                return f"Error executing tool: {str(e)}"
        
        return wrapper
    

    def run(self, transport: str = "stdio"):
        if not self.flight_server:
            self.start_flight_server()
        try:
            self.mcp.run(transport=transport)
        except Exception:
            raise
    
    def stop(self):
        pass


if __name__ == "__main__":
    print("FlightMCPServer class created")

1

Define the FlightMCPServer class that bridges MCP and Flight protocols.

2

Initialize with server name and Flight port (8818 by default).

3

Create placeholders for Flight server and client.

4

Track registered UDXFs and their exchange functions.

5

Start the Flight server that executes UDXFs.

6

Create Flight server with the specified port and registered UDXFs.

7

Begin serving Flight requests.

8

Get a client for calling UDXFs.

9

Create curried exchange functions for each UDXF.

10

Wrap a UDXF as an MCP tool that Claude can call.

11

Extract the UDXF command and set the tool name.

12

Register the UDXF.

13

Start Flight server if not already running.

14

Get the exchange function for this UDXF.

15

Use default output mapper if none is provided (converts DataFrame to string).

16

Define the async wrapper function that MCP calls.

17

Map MCP inputs to Xorq table using the input mapper.

18

Execute the UDXF via Flight protocol.

19

Format results using output mapper.

20

Run the MCP server on stdio transport (Claude Desktop uses stdin/stdout).

21

Test that the class definition works.

Changes

This diff shows the new code you’re adding to create the FlightMCPServer class. You’re importing necessary libraries and defining a class with methods to start Flight servers, register UDXFs as MCP tools, and handle communication between protocols.

+ from typing import Callable, Optional
+ import toolz
+ from mcp.server.fastmcp import FastMCP
+ from xorq.flight import FlightServer, FlightUrl
+ 
+ class FlightMCPServer:
+     def __init__(
+         self,
+         name: str,
+         flight_port: int = 8818,
+     ):
+         self.name = name
+         self.port = flight_port
+         self.mcp = FastMCP(name)
+         
+         self.flight_server = None
+         self.client = None
+         
+         self.udxfs = {}
+         self.schemas = {}
+         self.exchange_functions = {}
+     
+     def start_flight_server(self) -> FlightServer:
+         if self.flight_server:
+             return self.flight_server
+         
+         try:
+             self.flight_server = FlightServer(
+                 FlightUrl(port=self.port), 
+                 exchangers=list(self.udxfs.values())
+             )
+             
+             self.flight_server.serve()
+             self.client = self.flight_server.client
+             
+             for udxf_name, udxf in self.udxfs.items():
+                 self.exchange_functions[udxf_name] = toolz.curry(
+                     self.client.do_exchange, udxf.command
+                 )
+             
+             return self.flight_server
+         except Exception:
+             raise
+     
+     def create_mcp_tool(
+         self,
+         udxf,
+         input_mapper: Callable,
+         tool_name: Optional[str] = None,
+         description: Optional[str] = None,
+         output_mapper: Optional[Callable] = None,
+     ):
+         udxf_command = udxf.command
+         tool_name = tool_name or udxf_command
+         
+         self.udxfs[udxf_command] = udxf
+         
+         if not self.flight_server:
+             self.start_flight_server()
+         
+         do_exchange = self.exchange_functions.get(udxf_command)
+         
+         if output_mapper is None:
+             def default_output_mapper(result_df):
+                 if len(result_df) > 0:
+                     return result_df.to_string()
+                 return "No results"
+             
+             actual_output_mapper = default_output_mapper
+         else:
+             actual_output_mapper = output_mapper
+         
+         @self.mcp.tool(name=tool_name, description=description)
+         async def wrapper(**kwargs):
+             try:
+                 input_data = input_mapper(**kwargs)
+                 
+                 _, result = do_exchange(input_data.to_pyarrow_batches())
+                 result_df = result.read_pandas()
+                 
+                 output = actual_output_mapper(result_df)
+                 return output
+             except Exception as e:
+                 return f"Error executing tool: {str(e)}"
+         
+         return wrapper
+     
+     def run(self, transport: str = "stdio"):
+         if not self.flight_server:
+             self.start_flight_server()
+         try:
+             self.mcp.run(transport=transport)
+         except Exception:
+             raise
+     
+     def stop(self):
+         pass
+ 
+ if __name__ == "__main__":
+     print("FlightMCPServer class created")

Run the file to verify the class definition:

python sentiment_mcp_server.py

You’ll see:

FlightMCPServer class created

What just happened? You created a class that acts as a protocol translator. When Claude calls an MCP tool, FlightMCPServer receives the request, converts inputs to Xorq tables, calls the UDXF via Flight, and formats results back to Claude. The stdio transport means communication happens through standard input/output, which is how Claude Desktop connects to MCP servers.

Think of it this way: FlightMCPServer is like a bilingual interpreter. Claude speaks MCP (asking for tool executions), Xorq speaks Flight (executing UDXFs), and FlightMCPServer translates between them.

Build a sentiment analysis UDXF

Now you’ll create a function that analyzes text sentiment using OpenAI.

Update your sentiment_mcp_server.py file:

Complete code

This version adds the sentiment analysis UDXF to your file. It includes the OpenAI client setup, the analyze_sentiment function that calls GPT-3.5-turbo, schema definitions for input/output validation, and wraps everything as a UDXF using make_udxf.

from typing import Callable, Optional
import toolz
from mcp.server.fastmcp import FastMCP
from xorq.flight import FlightServer, FlightUrl

import functools
import os
from openai import OpenAI
import pandas as pd
import xorq.api as xo
from xorq.flight.exchanger import make_udxf

class FlightMCPServer:
    def __init__(
        self,
        name: str,
        flight_port: int = 8818,
    ):
        self.name = name
        self.port = flight_port
        self.mcp = FastMCP(name)
        
        self.flight_server = None
        self.client = None
        
        self.udxfs = {}
        self.schemas = {}
        self.exchange_functions = {}
    
    def start_flight_server(self) -> FlightServer:
        if self.flight_server:
            return self.flight_server
        
        try:
            self.flight_server = FlightServer(
                FlightUrl(port=self.port), 
                exchangers=list(self.udxfs.values())
            )
            
            self.flight_server.serve()
            self.client = self.flight_server.client
            
            for udxf_name, udxf in self.udxfs.items():
                self.exchange_functions[udxf_name] = toolz.curry(
                    self.client.do_exchange, udxf.command
                )
            
            return self.flight_server
        except Exception:
            raise
    
    def create_mcp_tool(
        self,
        udxf,
        input_mapper: Callable,
        tool_name: Optional[str] = None,
        description: Optional[str] = None,
        output_mapper: Optional[Callable] = None,
    ):
        udxf_command = udxf.command
        tool_name = tool_name or udxf_command
        
        self.udxfs[udxf_command] = udxf
        
        if not self.flight_server:
            self.start_flight_server()
        
        do_exchange = self.exchange_functions.get(udxf_command)
        
        if output_mapper is None:
            def default_output_mapper(result_df):
                if len(result_df) > 0:
                    return result_df.to_string()
                return "No results"
            
            actual_output_mapper = default_output_mapper
        else:
            actual_output_mapper = output_mapper
        
        @self.mcp.tool(name=tool_name, description=description)
        async def wrapper(**kwargs):
            try:
                input_data = input_mapper(**kwargs)
                
                _, result = do_exchange(input_data.to_pyarrow_batches())
                result_df = result.read_pandas()
                
                output = actual_output_mapper(result_df)
                return output
            except Exception as e:
                return f"Error executing tool: {str(e)}"
        
        return wrapper
    
    def run(self, transport: str = "stdio"):
        if not self.flight_server:
            self.start_flight_server()
        try:
            self.mcp.run(transport=transport)
        except Exception:
            raise
    
    def stop(self):
        pass


@functools.cache
def get_client():
    return OpenAI(api_key=os.environ["OPENAI_API_KEY"])


def analyze_sentiment(df: pd.DataFrame) -> pd.DataFrame:
    text = df["text"].iloc[0]
    
    if not text or text.strip() == "":
        return pd.DataFrame({"sentiment": ["NEUTRAL"]})
    

    messages = [
        {
            "role": "system",
            "content": "You are a sentiment analyzer. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL."
        },
        {
            "role": "user",
            "content": f"Analyze the sentiment: {text}"
        }
    ]
    

    try:
        response = get_client().chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=10,
            temperature=0,
        )
        sentiment = response.choices[0].message.content.strip()
        return pd.DataFrame({"sentiment": [sentiment]})
    except Exception:
        return pd.DataFrame({"sentiment": ["ERROR"]})


schema_in = xo.schema({"text": str})
schema_out = xo.schema({"sentiment": str})


sentiment_udxf = make_udxf(
    analyze_sentiment,
    schema_in,
    schema_out,
    name="sentiment_analyzer"
)

if __name__ == "__main__":

    test_df = pd.DataFrame({"text": ["This is amazing!"]})
    result = analyze_sentiment(test_df)
    print("Testing sentiment function:")
    print(result)
    
    print("\nSentiment UDXF created")

1

Add imports for OpenAI, pandas, and Xorq components.

2

Create a cached OpenAI client for reuse.

3

Define the sentiment analysis function that takes a DataFrame and returns sentiment.

4

Build a prompt that asks for one-word sentiment classification.

5

Call OpenAI API with error handling, returning “ERROR” on failure.

6

Define input schema (requires “text” column) and output schema (returns “sentiment” column).

7

Wrap the function as a UDXF that Xorq can execute via Flight protocol.

Changes

This diff shows the additions for sentiment analysis. You’re adding imports for OpenAI and pandas, creating a cached client, defining the sentiment analysis function with schema validation, and wrapping it as a UDXF that can be called via Flight protocol.

 from typing import Callable, Optional
 import toolz
 from mcp.server.fastmcp import FastMCP
 from xorq.flight import FlightServer, FlightUrl
+ import functools
+ import os
+ from openai import OpenAI
+ import pandas as pd
+ import xorq.api as xo
+ from xorq.flight.exchanger import make_udxf
 
 class FlightMCPServer:
     # ... (class definition unchanged)
 
+ @functools.cache
+ def get_client():
+     return OpenAI(api_key=os.environ["OPENAI_API_KEY"])
+ 
+ def analyze_sentiment(df: pd.DataFrame) -> pd.DataFrame:
+     text = df["text"].iloc[0]
+     
+     if not text or text.strip() == "":
+         return pd.DataFrame({"sentiment": ["NEUTRAL"]})
+     
+     messages = [
+         {
+             "role": "system",
+             "content": "You are a sentiment analyzer. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL."
+         },
+         {
+             "role": "user",
+             "content": f"Analyze the sentiment: {text}"
+         }
+     ]
+     
+     try:
+         response = get_client().chat.completions.create(
+             model="gpt-3.5-turbo",
+             messages=messages,
+             max_tokens=10,
+             temperature=0,
+         )
+         sentiment = response.choices[0].message.content.strip()
+         return pd.DataFrame({"sentiment": [sentiment]})
+     except Exception:
+         return pd.DataFrame({"sentiment": ["ERROR"]})
+ 
+ schema_in = xo.schema({"text": str})
+ schema_out = xo.schema({"sentiment": str})
+ 
+ sentiment_udxf = make_udxf(
+     analyze_sentiment,
+     schema_in,
+     schema_out,
+     name="sentiment_analyzer"
+ )
 
 if __name__ == "__main__":
+     # Test sentiment function
+     test_df = pd.DataFrame({"text": ["This is amazing!"]})
+     result = analyze_sentiment(test_df)
+     print("Testing sentiment function:")
+     print(result)
+     
+     print("\nSentiment UDXF created")

Run the file:

python sentiment_mcp_server.py

You’ll see:

Testing sentiment function:
  sentiment
0  POSITIVE

Sentiment UDXF created

Note

The function signature is crucial. It takes a pandas DataFrame and returns a pandas DataFrame. This contract lets Xorq validate schemas and handle data flow. If your input doesn’t have a “text” column, then Xorq catches the error before calling OpenAI.

Understanding the flow: Claude provides text → Input mapper creates DataFrame with “text” column → UDXF calls OpenAI → Returns DataFrame with “sentiment” column → Output mapper formats for Claude.

Create input and output mappers

Claude sends arguments as dictionaries. Your UDXF expects DataFrames. Mapper functions translate between these formats.

You need two mappers: one converts Claude’s text argument into a DataFrame for the UDXF, and one formats the UDXF’s DataFrame results into a dictionary Claude can display.

Update your sentiment_mcp_server.py file:

Complete code

This version adds mapper functions that translate between Claude’s tool arguments and Xorq’s table format. The input mapper converts text strings into DataFrames with proper schema validation. The output mapper formats UDXF results as dictionaries with human-readable interpretations for Claude.

from typing import Callable, Optional
import toolz
from mcp.server.fastmcp import FastMCP
from xorq.flight import FlightServer, FlightUrl
import functools
import os
from openai import OpenAI
import pandas as pd
import xorq.api as xo
from xorq.flight.exchanger import make_udxf

class FlightMCPServer:
    # ... (class definition unchanged)
    
    def __init__(
        self,
        name: str,
        flight_port: int = 8818,
    ):
        self.name = name
        self.port = flight_port
        self.mcp = FastMCP(name)
        
        self.flight_server = None
        self.client = None
        
        self.udxfs = {}
        self.schemas = {}
        self.exchange_functions = {}
    
    def start_flight_server(self) -> FlightServer:
        if self.flight_server:
            return self.flight_server
        
        try:
            self.flight_server = FlightServer(
                FlightUrl(port=self.port), 
                exchangers=list(self.udxfs.values())
            )
            
            self.flight_server.serve()
            self.client = self.flight_server.client
            
            for udxf_name, udxf in self.udxfs.items():
                self.exchange_functions[udxf_name] = toolz.curry(
                    self.client.do_exchange, udxf.command
                )
            
            return self.flight_server
        except Exception:
            raise
    
    def create_mcp_tool(
        self,
        udxf,
        input_mapper: Callable,
        tool_name: Optional[str] = None,
        description: Optional[str] = None,
        output_mapper: Optional[Callable] = None,
    ):
        udxf_command = udxf.command
        tool_name = tool_name or udxf_command
        
        self.udxfs[udxf_command] = udxf
        
        if not self.flight_server:
            self.start_flight_server()
        
        do_exchange = self.exchange_functions.get(udxf_command)
        
        if output_mapper is None:
            def default_output_mapper(result_df):
                if len(result_df) > 0:
                    return result_df.to_string()
                return "No results"
            
            actual_output_mapper = default_output_mapper
        else:
            actual_output_mapper = output_mapper
        
        @self.mcp.tool(name=tool_name, description=description)
        async def wrapper(**kwargs):
            try:
                input_data = input_mapper(**kwargs)
                
                _, result = do_exchange(input_data.to_pyarrow_batches())
                result_df = result.read_pandas()
                
                output = actual_output_mapper(result_df)
                return output
            except Exception as e:
                return f"Error executing tool: {str(e)}"
        
        return wrapper
    
    def run(self, transport: str = "stdio"):
        if not self.flight_server:
            self.start_flight_server()
        try:
            self.mcp.run(transport=transport)
        except Exception:
            raise
    
    def stop(self):
        pass

@functools.cache
def get_client():
    return OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def analyze_sentiment(df: pd.DataFrame) -> pd.DataFrame:
    text = df["text"].iloc[0]
    
    if not text or text.strip() == "":
        return pd.DataFrame({"sentiment": ["NEUTRAL"]})
    
    messages = [
        {
            "role": "system",
            "content": "You are a sentiment analyzer. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL."
        },
        {
            "role": "user",
            "content": f"Analyze the sentiment: {text}"
        }
    ]
    
    try:
        response = get_client().chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=10,
            temperature=0,
        )
        sentiment = response.choices[0].message.content.strip()
        return pd.DataFrame({"sentiment": [sentiment]})
    except Exception:
        return pd.DataFrame({"sentiment": ["ERROR"]})

schema_in = xo.schema({"text": str})
schema_out = xo.schema({"sentiment": str})

sentiment_udxf = make_udxf(
    analyze_sentiment,
    schema_in,
    schema_out,
    name="sentiment_analyzer"
)


def sentiment_input_mapper(**kwargs):

    text = kwargs.get("text", "")
    

    return xo.memtable({"text": [text]}, schema=schema_in)


def sentiment_output_mapper(result_df):

    if len(result_df) == 0:
        return {"sentiment": "NO_RESULT", "text": "No data to analyze"}
    

    sentiment = result_df["sentiment"].iloc[0]
    

    return {
        "sentiment": sentiment,
        "interpretation": f"The text sentiment is {sentiment}"
    }

if __name__ == "__main__":

    print("Testing input mapper:")
    test_input = sentiment_input_mapper(text="This is amazing!")
    print(test_input.execute())
    
    print("\nTesting output mapper:")
    test_output = sentiment_output_mapper(pd.DataFrame({"sentiment": ["POSITIVE"]}))
    print(test_output)
    
    print("\nMappers created and tested")

1

Define the input mapper that converts MCP tool arguments to Xorq tables.

2

Extract the “text” argument from Claude’s tool call.

3

Create a single-row table with the text, matching the UDXF’s expected schema.

4

Define the output mapper that formats UDXF results for Claude.

5

Handle empty results gracefully.

6

Extract the sentiment from the result DataFrame.

7

Return a dictionary with sentiment and human-readable interpretation.

8

Test both mappers to verify data transformation works correctly.

Changes

This diff adds the two mapper functions. The input mapper extracts text from Claude’s tool call and creates a Xorq table. The output mapper transforms the UDXF’s DataFrame result into a structured dictionary with sentiment and interpretation fields that Claude can present conversationally.

 sentiment_udxf = make_udxf(
     analyze_sentiment,
     schema_in,
     schema_out,
     name="sentiment_analyzer"
 )
 
+ def sentiment_input_mapper(**kwargs):
+     text = kwargs.get("text", "")
+     return xo.memtable({"text": [text]}, schema=schema_in)
+ 
+ def sentiment_output_mapper(result_df):
+     if len(result_df) == 0:
+         return {"sentiment": "NO_RESULT", "text": "No data to analyze"}
+     
+     sentiment = result_df["sentiment"].iloc[0]
+     
+     return {
+         "sentiment": sentiment,
+         "interpretation": f"The text sentiment is {sentiment}"
+     }
 
 if __name__ == "__main__":
+     # Test mappers
+     print("Testing input mapper:")
+     test_input = sentiment_input_mapper(text="This is amazing!")
+     print(test_input.execute())
+     
+     print("\nTesting output mapper:")
+     test_output = sentiment_output_mapper(pd.DataFrame({"sentiment": ["POSITIVE"]}))
+     print(test_output)
+     
+     print("\nMappers created and tested")

Run the file:

python sentiment_mcp_server.py

You’ll see:

Testing input mapper:
            text
0  This is amazing!

Testing output mapper:
{'sentiment': 'POSITIVE', 'interpretation': 'The text sentiment is POSITIVE'}

Mappers created and tested

What does success look like? Your mappers translate cleanly between formats. Claude sends {"text": "..."} → Input mapper creates a Xorq table → UDXF processes → Output mapper returns {"sentiment": "...", "interpretation": "..."} → Claude displays the result conversationally.

Most teams find that good output formatting makes a huge difference in user experience. Instead of raw DataFrames, Claude gets structured dictionaries with descriptive field names.

Wrap the UDXF as an MCP tool

Now you’ll connect all the pieces and create an MCP tool.

Update your sentiment_mcp_server.py file:

Complete code

This is the final working version of your MCP server. It includes the FlightMCPServer class, the sentiment UDXF, mapper functions, and the tool registration that connects everything. The if __name__ block handles both testing (default) and production mode (with --run flag).

from typing import Callable, Optional
import sys
import toolz
from mcp.server.fastmcp import FastMCP
from xorq.flight import FlightServer, FlightUrl
import functools
import os
from openai import OpenAI
import pandas as pd
import xorq.api as xo
from xorq.flight.exchanger import make_udxf

class FlightMCPServer:
    def __init__(
        self,
        name: str,
        flight_port: int = 8818,
    ):
        self.name = name
        self.port = flight_port
        self.mcp = FastMCP(name)
        
        self.flight_server = None
        self.client = None
        
        self.udxfs = {}
        self.schemas = {}
        self.exchange_functions = {}
    
    def start_flight_server(self) -> FlightServer:
        if self.flight_server:
            return self.flight_server
        
        try:
            self.flight_server = FlightServer(
                FlightUrl(port=self.port), 
                exchangers=list(self.udxfs.values())
            )
            
            self.flight_server.serve()
            self.client = self.flight_server.client
            
            for udxf_name, udxf in self.udxfs.items():
                self.exchange_functions[udxf_name] = toolz.curry(
                    self.client.do_exchange, udxf.command
                )
            
            return self.flight_server
        except Exception:
            raise
    
    def create_mcp_tool(
        self,
        udxf,
        input_mapper: Callable,
        tool_name: Optional[str] = None,
        description: Optional[str] = None,
        output_mapper: Optional[Callable] = None,
    ):
        udxf_command = udxf.command
        tool_name = tool_name or udxf_command
        
        self.udxfs[udxf_command] = udxf
        
        if not self.flight_server:
            self.start_flight_server()
        
        do_exchange = self.exchange_functions.get(udxf_command)
        
        if output_mapper is None:
            def default_output_mapper(result_df):
                if len(result_df) > 0:
                    return result_df.to_string()
                return "No results"
            
            actual_output_mapper = default_output_mapper
        else:
            actual_output_mapper = output_mapper
        
        @self.mcp.tool(name=tool_name, description=description)
        async def wrapper(**kwargs):
            try:
                input_data = input_mapper(**kwargs)
                
                _, result = do_exchange(input_data.to_pyarrow_batches())
                result_df = result.read_pandas()
                
                output = actual_output_mapper(result_df)
                return output
            except Exception as e:
                return f"Error executing tool: {str(e)}"
        
        return wrapper
    
    def run(self, transport: str = "stdio"):
        if not self.flight_server:
            self.start_flight_server()
        try:
            self.mcp.run(transport=transport)
        except Exception:
            raise
    
    def stop(self):
        pass

@functools.cache
def get_client():
    return OpenAI(api_key=os.environ["OPENAI_API_KEY"])

def analyze_sentiment(df: pd.DataFrame) -> pd.DataFrame:
    text = df["text"].iloc[0]
    
    if not text or text.strip() == "":
        return pd.DataFrame({"sentiment": ["NEUTRAL"]})
    
    messages = [
        {
            "role": "system",
            "content": "You are a sentiment analyzer. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL."
        },
        {
            "role": "user",
            "content": f"Analyze the sentiment: {text}"
        }
    ]
    
    try:
        response = get_client().chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=10,
            temperature=0,
        )
        sentiment = response.choices[0].message.content.strip()
        return pd.DataFrame({"sentiment": [sentiment]})
    except Exception:
        return pd.DataFrame({"sentiment": ["ERROR"]})

schema_in = xo.schema({"text": str})
schema_out = xo.schema({"sentiment": str})

sentiment_udxf = make_udxf(
    analyze_sentiment,
    schema_in,
    schema_out,
    name="sentiment_analyzer"
)

def sentiment_input_mapper(**kwargs):
    text = kwargs.get("text", "")
    return xo.memtable({"text": [text]}, schema=schema_in)

def sentiment_output_mapper(result_df):
    if len(result_df) == 0:
        return {"sentiment": "NO_RESULT", "text": "No data to analyze"}
    
    sentiment = result_df["sentiment"].iloc[0]
    
    return {
        "sentiment": sentiment,
        "interpretation": f"The text sentiment is {sentiment}"
    }


mcp_server = FlightMCPServer("xorq-sentiment")


mcp_server.create_mcp_tool(
    sentiment_udxf,
    input_mapper=sentiment_input_mapper,
    tool_name="analyze_sentiment",
    description="Analyze the sentiment of text. Returns POSITIVE, NEGATIVE, or NEUTRAL.",
    output_mapper=sentiment_output_mapper
)


if __name__ == "__main__":

    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "--run":
        try:
            mcp_server.run(transport="stdio")
        except Exception:
            sys.exit(1)
    else:
        print("MCP tool registered: analyze_sentiment")
        print("\nReady for Claude Desktop integration")
        print("Run with: python sentiment_mcp_server.py --run")

1

Create a FlightMCPServer instance named “xorq-sentiment”.

2

Register the sentiment UDXF as an MCP tool with custom mappers and a clear description.

3

Add command-line handling for testing vs production use.

4

When run with --run flag, start the MCP server for Claude Desktop. Otherwise, just print status.

Changes

This diff shows the final integration code. You’re creating a FlightMCPServer instance, registering the sentiment UDXF as an MCP tool with custom mappers and description, and adding command-line handling to support both testing and production modes.

 def sentiment_output_mapper(result_df):
     if len(result_df) == 0:
         return {"sentiment": "NO_RESULT", "text": "No data to analyze"}
     
     sentiment = result_df["sentiment"].iloc[0]
     
     return {
         "sentiment": sentiment,
         "interpretation": f"The text sentiment is {sentiment}"
     }
 
+ mcp_server = FlightMCPServer("xorq-sentiment")
+ 
+ mcp_server.create_mcp_tool(
+     sentiment_udxf,
+     input_mapper=sentiment_input_mapper,
+     tool_name="analyze_sentiment",
+     description="Analyze the sentiment of text. Returns POSITIVE, NEGATIVE, or NEUTRAL.",
+     output_mapper=sentiment_output_mapper
+ )
 
 if __name__ == "__main__":
+     import sys
+     if len(sys.argv) > 1 and sys.argv[1] == "--run":
+         try:
+             mcp_server.run(transport="stdio")
+         except Exception:
+             sys.exit(1)
+     else:
+         print("FlightMCPServer class created")
+         print("Sentiment UDXF created")
+         print("Mappers created")
+         print("MCP tool registered: analyze_sentiment")
+         print("\nReady for Claude Desktop integration")
+         print("Run with: python sentiment_mcp_server.py --run")

Run the file to verify everything is set up:

python sentiment_mcp_server.py

The output shows:

[01/05/26 08:56:22] INFO     Running action healthcheck
                    INFO     doing action: healthcheck
                    INFO     done healthcheck
                    INFO     Flight server unavailable, sleeping 1 seconds
[01/05/26 08:56:23] INFO     Running action add-exchange
                    INFO     doing action: add-exchange
MCP tool registered: analyze_sentiment

Ready for Claude Desktop integration
Run with: python sentiment_mcp_server.py --run

The INFO messages show the Flight server starting up and registering your UDXF. This is normal.

What just happened? You created an MCP tool that Claude can discover and call. The tool has a name (analyze_sentiment), a description (tells Claude what it does), input mapping (converts Claude’s arguments to Xorq format), and output mapping (formats results for Claude).

Understanding the registration: FlightMCPServer starts a Flight server in the background, registers your UDXF, and exposes it through the MCP protocol. When Claude calls analyze_sentiment, the server routes the request through Flight to your UDXF.

Configure Claude Desktop

Now you’ll tell Claude Desktop where to find your MCP tool. You need to edit a configuration file that Claude Desktop reads on startup.

Step 1: Find your script’s full path

You need the complete path to sentiment_mcp_server.py.

macOS / Linux

Open your terminal in the directory where sentiment_mcp_server.py is located, then run:

pwd

This shows your current directory path. Copy the output.

Example output:

/Users/yourname/projects/xorq-tutorials

Your full script path is this directory + the filename:

/Users/yourname/projects/xorq-tutorials/sentiment_mcp_server.py

Copy this full path - you’ll need it in Step 2.

Windows

Open your terminal (Command Prompt or PowerShell) in the directory where sentiment_mcp_server.py is located, then run:

cd

This shows your current directory path. Copy the output.

Example output:

C:\Users\yourname\projects\xorq-tutorials

Your full script path is this directory + the filename:

C:\Users\yourname\projects\xorq-tutorials\sentiment_mcp_server.py

Copy this full path - you’ll need it in Step 2.

Step 2: Find the config file

Claude Desktop reads its configuration from a specific file.

macOS / Linux

The config file is located at:

~/Library/Application Support/Claude/config.json

Open this file in a text editor. If the file doesn’t exist, then create it:

# Create the directory if it doesn't exist
mkdir -p ~/Library/Application\ Support/Claude

# Create the empty config file
touch ~/Library/Application\ Support/Claude/config.json

Then open it with your text editor:

open ~/Library/Application\ Support/Claude/config.json

Windows

The config file is located at:

%APPDATA%\Claude\claude_desktop_config.json

Open File Explorer and navigate to this file:

1. Press Win + R
2. Type: %APPDATA%\Claude
3. Press Enter

If you see claude_desktop_config.json, then open it in a text editor like VS Code or Notepad++.

If the file doesn’t exist:

1. Right-click in the folder
2. Select New → Text Document
3. Name it exactly: config.json
4. Open it in a text editor

Important: Make sure it’s named config.json, not claude_desktop_config.json.txt. You may need to show file extensions in File Explorer.

Step 3: Add your MCP server configuration

Copy this configuration into the file, replacing the example path with your actual path from Step 1.

macOS / Linux

Copy this into the config file:

{
  "mcpServers": {
    "xorq-sentiment": {
      "command": "python",
      "args": ["/YOUR/PATH/HERE/sentiment_mcp_server.py", "--run"]
    }
  }
}

Replace /YOUR/PATH/HERE/sentiment_mcp_server.py with your actual path from Step 1.

Example with real path:

{
  "mcpServers": {
    "xorq-sentiment": {
      "command": "python",
      "args": ["/Users/yourname/projects/xorq-tutorials/sentiment_mcp_server.py", "--run"]
    }
  }
}

Save the file.

Windows

Copy this into the config file:

{
  "mcpServers": {
    "xorq-sentiment": {
      "command": "python",
      "args": ["C:\\YOUR\\PATH\\HERE\\sentiment_mcp_server.py", "--run"]
    }
  }
}

Replace C:\\YOUR\\PATH\\HERE\\sentiment_mcp_server.py with your actual path from Step 1.

Important: Use double backslashes (\\) between folders, not single backslashes.

Example with real path:

{
  "mcpServers": {
    "xorq-sentiment": {
      "command": "python",
      "args": ["C:\\Users\\yourname\\projects\\xorq-tutorials\\sentiment_mcp_server.py", "--run"]
    }
  }
}

Save the file.

Step 4: Restart Claude Desktop

Important: You must fully quit and restart Claude Desktop.

On macOS:

1. Click Claude in the menu bar
2. Click Quit Claude
3. Reopen Claude Desktop from Applications

On Windows:

1. Right-click the Claude icon in the system tray
2. Click Quit
3. Reopen Claude Desktop from the Start menu

After restarting, look for a tool indicator (wrench icon or tool count) in Claude Desktop. This confirms your tool is connected.

Troubleshooting: Tool not appearing?

If you don’t see the tool indicator after restarting:

• Open the config file again
• Verify the path is absolute (starts with / on macOS/Linux or C:\ on Windows)
• On Windows, verify you used double backslashes: \\
• Verify the --run flag is in the args array
• Check for typos in "xorq-sentiment" (must match FlightMCPServer("xorq-sentiment"))

Test your script works:

Run the script from your terminal to verify it starts without errors and registers the tool correctly.

macOS / Linux

Run the script using the absolute path to your file:

python /Users/yourname/projects/xorq-tutorials/sentiment_mcp_server.py

Should show: MCP tool registered: analyze_sentiment

Windows

Run the script using the absolute path to your file:

python C:\Users\yourname\projects\xorq-tutorials\sentiment_mcp_server.py

Should show: MCP tool registered: analyze_sentiment

Check Claude Desktop logs:

• macOS: ~/Library/Logs/Claude/mcp.log
• Windows: %APPDATA%\Claude\logs\mcp.log

Look for errors related to your MCP server. :::

Test with Claude Desktop

Once Claude Desktop is configured and restarted, you can test your tool:

Open Claude Desktop and look for the tool indicator (usually a wrench icon or tool count).

Ask Claude to use your tool:

"Can you analyze the sentiment of: This product exceeded all my expectations!"

Claude calls your tool and you see the response:

I've analyzed the sentiment using the sentiment analyzer tool.

Result: POSITIVE
The text sentiment is POSITIVE

The review expresses strong satisfaction and excitement about the product.

What just happened? Claude detected that your request needed sentiment analysis, called the analyze_sentiment tool through MCP, your FlightMCPServer routed it to the UDXF via Flight, OpenAI processed the text, and results flowed back to Claude for presentation.

Understanding the flow: Your question → Claude’s reasoning → MCP tool call → FlightMCPServer → Flight → UDXF → OpenAI API → Results back through the chain → Claude’s response.

Debugging tool calls

If Claude doesn’t call your tool, then try:

1. Explicitly mention “use the analyze_sentiment tool”.
2. Check Claude Desktop logs for connection errors.
3. Run python sentiment_mcp_server.py to verify the script works without errors
4. Verify the --run flag is in your config.

Extend with multiple tools

Now that you have one tool working, you might wonder: can you expose multiple UDXFs as different tools? Yes. The pattern is identical for each additional tool you want to expose.

Create a file called multi_tool_mcp_server.py with two tools:

from typing import Callable, Optional
import sys
import toolz
from mcp.server.fastmcp import FastMCP
from xorq.flight import FlightServer, FlightUrl
import functools
import os
from openai import OpenAI
import pandas as pd
import xorq.api as xo
from xorq.flight.exchanger import make_udxf

# FlightMCPServer class (same as before)
class FlightMCPServer:
    def __init__(self, name: str, flight_port: int = 8818):
        self.name = name
        self.port = flight_port
        self.mcp = FastMCP(name)
        self.flight_server = None
        self.client = None
        self.udxfs = {}
        self.schemas = {}
        self.exchange_functions = {}
    
    def start_flight_server(self) -> FlightServer:
        if self.flight_server:
            return self.flight_server
        
        try:
            self.flight_server = FlightServer(
                FlightUrl(port=self.port), 
                exchangers=list(self.udxfs.values())
            )
            self.flight_server.serve()
            self.client = self.flight_server.client
            
            for udxf_name, udxf in self.udxfs.items():
                self.exchange_functions[udxf_name] = toolz.curry(
                    self.client.do_exchange, udxf.command
                )
            
            return self.flight_server
        except Exception:
            raise
    
    def create_mcp_tool(
        self,
        udxf,
        input_mapper: Callable,
        tool_name: Optional[str] = None,
        description: Optional[str] = None,
        output_mapper: Optional[Callable] = None,
    ):
        udxf_command = udxf.command
        tool_name = tool_name or udxf_command
        self.udxfs[udxf_command] = udxf
        
        if not self.flight_server:
            self.start_flight_server()
        
        do_exchange = self.exchange_functions.get(udxf_command)
        
        if output_mapper is None:
            def default_output_mapper(result_df):
                if len(result_df) > 0:
                    return result_df.to_string()
                return "No results"
            actual_output_mapper = default_output_mapper
        else:
            actual_output_mapper = output_mapper
        
        @self.mcp.tool(name=tool_name, description=description)
        async def wrapper(**kwargs):
            try:
                input_data = input_mapper(**kwargs)
                _, result = do_exchange(input_data.to_pyarrow_batches())
                result_df = result.read_pandas()
                output = actual_output_mapper(result_df)
                return output
            except Exception as e:
                return f"Error executing tool: {str(e)}"
        
        return wrapper
    
    def run(self, transport: str = "stdio"):
        if not self.flight_server:
            self.start_flight_server()
        try:
            self.mcp.run(transport=transport)
        except Exception:
            raise
    
    def stop(self):
        pass

# OpenAI client
@functools.cache
def get_client():
    return OpenAI(api_key=os.environ["OPENAI_API_KEY"])

# TOOL 1: Sentiment analyzer
def analyze_sentiment(df: pd.DataFrame) -> pd.DataFrame:
    text = df["text"].iloc[0]
    
    if not text or text.strip() == "":
        return pd.DataFrame({"sentiment": ["NEUTRAL"]})
    
    messages = [
        {
            "role": "system",
            "content": "You are a sentiment analyzer. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL."
        },
        {
            "role": "user",
            "content": f"Analyze the sentiment: {text}"
        }
    ]
    
    try:
        response = get_client().chat.completions.create(
            model="gpt-3.5-turbo",
            messages=messages,
            max_tokens=10,
            temperature=0,
        )
        sentiment = response.choices[0].message.content.strip()
        return pd.DataFrame({"sentiment": [sentiment]})
    except Exception:
        return pd.DataFrame({"sentiment": ["ERROR"]})

sentiment_schema_in = xo.schema({"text": str})
sentiment_schema_out = xo.schema({"sentiment": str})

sentiment_udxf = make_udxf(
    analyze_sentiment,
    sentiment_schema_in,
    sentiment_schema_out,
    name="sentiment_analyzer"
)

def sentiment_input_mapper(**kwargs):
    text = kwargs.get("text", "")
    return xo.memtable({"text": [text]}, schema=sentiment_schema_in)

def sentiment_output_mapper(result_df):
    if len(result_df) == 0:
        return {"sentiment": "NO_RESULT", "text": "No data to analyze"}
    
    sentiment = result_df["sentiment"].iloc[0]
    return {
        "sentiment": sentiment,
        "interpretation": f"The text sentiment is {sentiment}"
    }

# TOOL 2: Text length counter
def count_text_length(df: pd.DataFrame) -> pd.DataFrame:
    text = df["text"].iloc[0]
    length = len(text) if text else 0
    word_count = len(text.split()) if text else 0
    
    return pd.DataFrame({
        "character_count": [length],
        "word_count": [word_count]
    })

length_schema_in = xo.schema({"text": str})
length_schema_out = xo.schema({"character_count": int, "word_count": int})

length_udxf = make_udxf(
    count_text_length,
    length_schema_in,
    length_schema_out,
    name="text_length_counter"
)

def length_input_mapper(**kwargs):
    text = kwargs.get("text", "")
    return xo.memtable({"text": [text]}, schema=length_schema_in)

def length_output_mapper(result_df):
    if len(result_df) == 0:
        return {"error": "No data"}
    
    return {
        "character_count": int(result_df["character_count"].iloc[0]),
        "word_count": int(result_df["word_count"].iloc[0]),
        "summary": f"{result_df['word_count'].iloc[0]} words, {result_df['character_count'].iloc[0]} characters"
    }

# Create MCP server and register both tools
mcp_server = FlightMCPServer("xorq-multi-tools")

mcp_server.create_mcp_tool(
    sentiment_udxf,
    input_mapper=sentiment_input_mapper,
    tool_name="analyze_sentiment",
    description="Analyze the sentiment of text. Returns POSITIVE, NEGATIVE, or NEUTRAL.",
    output_mapper=sentiment_output_mapper
)

mcp_server.create_mcp_tool(
    length_udxf,
    input_mapper=length_input_mapper,
    tool_name="count_text_length",
    description="Count characters and words in text",
    output_mapper=length_output_mapper
)

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--run":
        try:
            mcp_server.run(transport="stdio")
        except Exception:
            sys.exit(1)
    else:
        # Test both UDXFs
        print("Testing sentiment analyzer:")
        test_sentiment = analyze_sentiment(pd.DataFrame({"text": ["This is amazing!"]}))
        print(test_sentiment)
        
        print("\nTesting length counter:")
        test_length = count_text_length(pd.DataFrame({"text": ["Hello world from Xorq!"]}))
        print(test_length)
        
        print("\nBoth tools registered")
        print("Run with: python multi_tool_mcp_server.py --run")

Run the file to test:

python multi_tool_mcp_server.py

The output shows:

Testing sentiment analyzer:
  sentiment
0  POSITIVE

Testing length counter:
   character_count  word_count
0               23           4

Both tools registered
Run with: python multi_tool_mcp_server.py --run

Update Claude Desktop config:

macOS / Linux

{
  "mcpServers": {
    "xorq-multi-tools": {
      "command": "python",
      "args": ["/absolute/path/to/multi_tool_mcp_server.py", "--run"]
    }
  }
}

Windows

{
  "mcpServers": {
    "xorq-multi-tools": {
      "command": "python",
      "args": ["C:\\absolute\\path\\to\\multi_tool_mcp_server.py", "--run"]
    }
  }
}

Test in Claude Desktop:

"Count the words in: This is a test message for the counter"

Claude calls count_text_length and responds:

The text contains:

- 9 words
- 44 characters

What does success look like? You’ve exposed two different UDXFs as separate tools. Claude intelligently selects which tool to use based on your request. Ask about sentiment, and Claude calls analyze_sentiment. Ask about word count, and Claude calls count_text_length.

The pattern scales: create UDXF → define mappers → register with create_mcp_tool(). Each tool operates independently, and Claude selects the appropriate one for each request.

What you learned

You’ve learned how to expose Xorq UDXFs as tools Claude can call through MCP. Here’s what you accomplished:

• Understood MCP (Model Context Protocol) and its role in Claude integrations
• Created FlightMCPServer to bridge MCP and Flight protocols
• Built a sentiment analysis UDXF using OpenAI
• Implemented input and output mappers for data transformation
• Registered UDXFs as MCP tools with clear descriptions
• Configured Claude Desktop to discover and use your tools
• Tested the integration with conversational queries
• Extended the pattern to multiple tools

The key insight? MCP enables natural language access to Xorq functions. You wrap your data processing logic once, and Claude can call it conversationally. The FlightMCPServer handles all protocol translation, letting you focus on building useful tools.

Next steps

Now that you know how to build MCP tools, continue learning:

• Add LLM analysis with UDXFs explains UDXF architecture and schema patterns in depth
• Call LLMs from expressions shows advanced LLM integration techniques
• Your first build covers versioning and deploying Xorq pipelines