Article

Build an Airbyte MCP Server for Claude Desktop

•

March 10, 2025

•

10 min read

MCP, or Model Context Protocol, has been gaining a lot of attention. Originally proposed by Anthropic, MCP is an open standard for AI agents to plug into tools and services such a Cursor, Claude etc. As we see more and more workloads shifting to AI agents, I wanted to explore how this might work with data movement and the APIs provided by the Airbyte platform.

Let’s take a example of checking the status of connections within Airbyte. MCP should allow me to create a service which interacts with the Airbyte APIs, and expose this as a server accessible by something like Claude Desktop.

‍

There are three primary components I need to create:

An MCP server which conforms to the MCP protocol
The logic in my server which knows how to communicate with Airbyte and retrieve connection status. I then need to expose this function to MCP
An MCP server configuration file for Claude Desktop

‍

It starts with a prompt

I’m using Cursor to generate the initial code for my server. Here is the prompt I used. This prompt defines the intent of what I want to build and clear implementation instructions on how to build it and what coding conventions I like to follow. In my experience, spending a little more time in prompt definition by providing detailed instructions is well worth it. I’m using Claude 3.7 for the model. It is by far the best I have found for code generation.


Create an MCP server that uses the airbyte to connect to a cloud workspace to perform a status check on sources. The server should:

2. Allow me to pass in the source name, or if no name is provided, return a list of all sources
3. Return the results of the status check with an informative message and helpful emoji
4. make a build version of the server so it can be used in Claude Desktop
5. Create a supporting claude_desktop_config.json file 

Important implementation details:
- Follow the guide here to implement the server: https://modelcontextprotocol.io/quickstart/server
- use fastmcp to implement the mcp https://github.com/modelcontextprotocol/python-sdk
- create a .env file to store workspace id, client id, client secret, and auth token required by airbyte
- implement token refresh when using Airbyte APIs
- create a claude_desktop_config.json
- Do not make assumptions about class names or method signatures
- make sure you use the correct mcp annotations like @mcp.tool()
- when creating the claude_desktop_config.json, use this doc as a guide: https://modelcontextprotocol.io/quickstart/user#for-claude-desktop-users
- Make all paths absolute in the claude_desktop_config.json. Use uv and python for executables. Here is an example. Strictly stick to this.
  - "command": ["python"],
    "args": [
    "--directory",
    "/ABSOLUTE/PATH/TO/PARENT/FOLDER/airbyte-status-checker",
    "run",
    "airbyte_status_checker.py"
    ],
  
  "args": [
  "--directory",
  "/ABSOLUTE/PATH/TO/PARENT/FOLDER/myapp”,
  "run",
  “myapp.py”
  ]

‍

Dependencies

Cursor should have created a requirements.txt for you.


fastmcp>=0.1.0
python-dotenv>=1.0.0
requests>=2.31.0

‍

I use uv for virtual environments, and can install my dependencies via:


uv pip install -r requirements.txt

‍

Setting up your server

We are going to be using FastMCP to define our server. FastMCP takes care a lot of the connection management, protocol compliance, and message routing that you need to conform to the MCP protocol. The most important part of your server will be defining a function which is exposed via the @mcp.tool() decorator. This tells apps like Claude Desktop that the function is accessible as a service. Think of it as a webhook or route in React app.


@mcp.tool()
async def check_airbyte_connection(connection_name: Optional[str] = None) -> Dict[str, Any]:

‍

MCP also allows you to further define the keyword or phrase your client app, like Claude, should listen for to know when you want to invoke your server. Here is an example from another project that defines the phrase as “ask airbyte”:


@mcp.tool()
async def ask_question(question: str) -> Dict[str, str]:
    """
    Ask a question about your Contract data - powered by Airbyte.
    
    Special commands:
    - "ask Airbyte:  " - Ask AI bot questions about Contracts
   
    """

‍

Airbyte Connection Status

Let’s initialize our server and set up all of the secrets to help us authenticate with Airbyte. As specified in the initial prompt, Claude has created the logic to retrieve secrets from a .env file.


import os
import json
import requests
import time
from typing import Optional, Dict, Any, List
from dotenv import load_dotenv
from fastmcp import FastMCP

# Create a FastMCP instance
mcp = FastMCP("Airbyte Connection Checker")

# Load environment variables
load_dotenv()

# Airbyte API configuration
API_BASE_URL = "https://api.airbyte.com/v1"
API_KEY = os.getenv("AIRBYTE_API_KEY")
WORKSPACE_ID = os.getenv("AIRBYTE_WORKSPACE_ID")
CLIENT_ID = os.getenv("AIRBYTE_CLIENT_ID")
CLIENT_SECRET = os.getenv("AIRBYTE_CLIENT_SECRET")

if not WORKSPACE_ID or not API_KEY:
    raise ValueError("Missing required Airbyte credentials in .env file")

# Headers for API requests
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

‍

Once I have my initialization complete, it is best practice to handle token refreshes for Airbyte prior to every API call. Airbyte uses a short lived token for security.


# Token refresh function
def refresh_airbyte_token():
    """Refresh the Airbyte API token"""
    try:
        refresh_token = os.getenv('AIRBYTE_API_KEY')
        if not refresh_token:
            raise ValueError("AIRBYTE_API_KEY not found in environment variables")

        base_url = API_BASE_URL.rstrip('/')
        response = requests.post(
            f'{base_url}/applications/token',
            data={
                'grant_type': 'refresh_token',
                'client_id': CLIENT_ID,
                'client_secret': CLIENT_SECRET,
                'refresh_token': refresh_token
            }
        )
        response.raise_for_status()

        new_token = response.json().get('access_token')
        if not new_token:
            raise ValueError("No access token in response")

        # Update global API_KEY and HEADERS
        global API_KEY, HEADERS
        API_KEY = new_token
        HEADERS = {
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json"
        }
        
        return new_token
    except Exception as e:
        print(f"Failed to refresh token: {str(e)}")
        raise

‍

Now, I need a few functions to actually fetch the connections and associated streams and status. In this example, I didn’t use the Airbyte client SDKs. My use case was pretty simple to work directly with the APIs.


def get_connections():
    """Get all connections in the workspace"""
    url = f"{API_BASE_URL}/connections"
    params = {"workspaceIds": WORKSPACE_ID}
    
    response = requests.get(url, headers=HEADERS, params=params)
    response.raise_for_status()
    
    return response.json().get("data", [])

def check_connection_status(connection_id):
    """Check the status of a connection"""
    url = f"{API_BASE_URL}/connections/get"
    
    payload = {
        "connectionId": connection_id
    }
    
    response = requests.post(url, headers=HEADERS, json=payload)
    response.raise_for_status()
    
    return response.json()

def get_connection_streams(connection_id):
    """Get the streams for a connection"""
    url = f"{API_BASE_URL}/connections/get"
    
    payload = {
        "connectionId": connection_id
    }
    
    response = requests.post(url, headers=HEADERS, json=payload)
    response.raise_for_status()
    
    connection_data = response.json()
    # Extract stream information from the connection data
    # This might need adjustment based on the actual API response structure
    streams = connection_data.get("syncCatalog", {}).get("streams", [])
    return [stream.get("stream", {}).get("name") for stream in streams if stream.get("config", {}).get("selected")]

‍

All that is left is to implement the check_airbyte_connection function which we defined as an MCP server endpoint via the @mcp.tool() decorator. Once I get the results from Airbyte, I return the results as a Dictionary object and let Claude Desktop handle the display. Returning your results as a Dictionary makes your server very portable between different tools.


@mcp.tool()
async def check_airbyte_connection(connection_name: Optional[str] = None) -> Dict[str, Any]:
    """
    Check the status of an Airbyte connection or list all connections.
    
    Args:
        connection_name: Name of the connection to check. If not provided, lists all connections.
        
    Returns:
        A dictionary with status information.
    """
    try:
        # Try to refresh the token if needed
        try:
            if CLIENT_ID and CLIENT_SECRET:
                refresh_airbyte_token()
        except Exception as e:
            print(f"Token refresh failed, continuing with existing token: {str(e)}")
        
        # Get all connections in the workspace
        connections = get_connections()
        
        if not connection_name:
            # If no connection name provided, return list of all connections
            connection_list = [
                {
                    "name": conn.get("name"), 
                    "id": conn.get("connectionId"), 
                    "status": "🟢 Active" if conn.get("status", "").lower() == "active" else "🔴 Inactive"
                } 
                for conn in connections
            ]
            
            return {
                "status": "success",
                "message": "📋 Here's a list of all connections",
                "connections": connection_list
            }
        else:
            # Find the connection by name
            connection = None
            for conn in connections:
                if conn.get("name", "").lower() == connection_name.lower():
                    connection = conn
                    break
            
            if not connection:
                return {
                    "status": "error",
                    "message": f"❌ Connection '{connection_name}' not found"
                }
            
            # Get connection details
            connection_id = connection.get("connectionId")
            connection_details = check_connection_status(connection_id)
            
            # Get streams for this connection
            streams = get_connection_streams(connection_id)
            
            status = connection.get("status", "")
            
            if status.lower() == "active":
                emoji = "✅"
                message = f"Connection '{connection_name}' is active"
            else:
                emoji = "❌"
                message = f"Connection '{connection_name}' is inactive"
            
            return {
                "status": status,
                "message": f"{emoji} {message}",
                "connection_name": connection_name,
                "connection_id": connection_id,
                "streams": streams,
                "details": connection_details
            }
                
    except Exception as e:
        return {
            "status": "error",
            "message": f"❌ Error: {str(e)}"
        }

‍

Once you’ve finished coding your server, add support to allow you to run your server from the command line. When debugging my server to check it did not have any syntax errors, I found it valuable to run the server to ensure it at least starts. You can’t invoke the MCP server, but you can confirm that your server starts correctly.


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')

‍

claude_desktop_config.json

Every AI tool will have a different way of configuring it to load your MCP Server. Claude Desktop uses a file called claude_desktop_config.json. You can find the location of it on your machine by opening Claude Desktop > Settings > Developer > Edit config. Please note that MCP only works via Claude Desktop, not the browser based version.

When running locally in the command line, we’ve set up our server to run locally via the following command.


uv run airbyte_status_checker.py

‍

I need to configure my config file to give it instructions on where to find this file and how to execute it locally. You can also have a remote MCP server accessible via http/s, but in my case I need to configure Claude with the exact path to my local server. You will need to change yours to match your path and what ever virtual environment you use.


{
  "mcpServers": {
    "airbyte-status-checker": {
      "command": "/Users/quintonwall/.local/bin/uv",
      "args": [
        "--directory",
        "/Users/quintonwall/code/airbyte-mcp-list-sources",
        "run",
        "airbyte_status_checker.py"
      ]
    }
  }
}

‍

Running your server

With everything coded, open up Claude Desktop and add your configuration via Settings > Developer > Edit Config, then restart Claude. If everything is configured correctly, you should see a small hammer icon.

If you don’t see this, your claude_desktop_config.json is not configured correctly. Double check your paths and make sure everything has absolute paths and is 100% correct.

Once up and running you can tap on the hammer icon to see information about all of your available servers

Typing in a prompt such as “Show me the status of my connections in Airbyte” will return a nicely formatted list of connections and status. Nice.

‍

Debugging

If you are still having trouble getting your server to be recognized, Claude outputs MCP server start up logs to ~/Library/Logs/Claude/mcp.log on a mac. Unfortunately, if your config file is not actually finding your server executable, Claude doesn’t currently print any information, you just have to manually confirm paths. Once Claude finds it (and you see the hammer icon), and perhaps has an error starting your server, the results will be written to a file called mcp.log

Once Claude has recognized your server correctly, and your server is up and running, any logging information you have configured in your server will be written to a log file with the name defined in your configuration file for your server. In my instance, this is airbyte-status-checker.

Summary

MCP is an exciting way to add tools and services to AI assistants. It’s still early, but the protocol is gaining a lot of traction amongst developers and startups. If you want to check out the complete code from this guide, you can grab it via GitHub and sign up for Airbyte Free Trial account to test it out.

‍

Limitless data movement with free Alpha and Beta connectors

Introducing: our Free Connector Program ->

The data movement infrastructure for the modern data teams.

Try a 14-day free trial