Forem: Airbyte

One-click MCP installs with npx

Quinton — Wed, 09 Jul 2025 19:55:42 +0000

A few weeks back I shipped the PyAirbyte MCP service for Airbyte. This service relies on OpenAI and Vector Embeddings to create an entire data pipeline with a single prompt. All you had to do was provide a source and destination from any of the hundreds of connectors available in the Airbyte Registry.

"create a data pipeline from source-postures to destination-data bricks"
_

The service has been getting heavy usage, which is always great. The biggest bit of feedback that I receive though, is "I wish it ran in Claude/Cline/Replit/etc".

The initial release was shipped as a remote MCP server running on Heroku. This implementation relies on the user passing in their OpenAI API key, which the server uses, along with the vector embed search, to determine how to generate the right code. All sounds great until you dig into the how different vendors implement the mcp.json spec. At the time of publishing the MCP, only Cursor allows the passing in of environment variables within mcp.json for remote servers. I wanted to ship v1 as fast as possible, so I decided to only support Cursor -- not a bad thing if a million developers already use it, right?

"pyairbyte-mcp": {
      "url": "https://pyairbyte-mcp-7b7b8566f2ce.herokuapp.com/mcp",
      "env": {
        "OPENAI_API_KEY": "your-openai-api-key"
      }
    }

NPX - the secret ingredient

After shipping v1, I wanted to find a way to easily allow developers to add the MCP server to whatever AI tool they wanted. I'm a big fan of Cline for my IDE and Claude Desktop as my AI Guide to the Galaxy. With a little digging, npx seemed to the secret ingredient.

NPX is a CLI tool for installing and managing dependencies of a host npm package on npmjs.com. Since NPX is a CLI tool, you can execute it as part of the mcp.json spec:

"fast-pyairbyte": {
      "command": "npx",
      "args": ["fast-pyairbyte"],
      "env": {
        "OPENAI_API_KEY": "your-key-here"
      }
    }

From my testing, all of the major apps that support MCP.json handle npx great. To the app it is just another local execution. To the user, it is a single step install. Here is the MCP running right within Claude Desktop.

Try it for yourself

The MCP is now an official package on npmjs.com, under the name fast-pyairbyte. Cursor still leads the way in terms of developer experience to install an MCP server with one click installs:

If you are using Cline, Claude, or pretty much any other tool, you can add Fast PyAirbyte by including the json snippet into the relevant config files:

For Cursor (.cursor/mcp.json):

{
  "mcpServers": {
    "fast-pyairbyte": {
      "command": "npx",
      "args": ["fast-pyairbyte"],
      "env": {
        "OPENAI_API_KEY": "your-openai-api-key-here"
      }
    }
  }
}

For Claude Desktop (~/.config/claude/claude_desktop_config.json):

{
  "mcpServers": {
    "fast-pyairbyte": {
      "command": "npx",
      "args": ["fast-pyairbyte"],
      "env": {
        "OPENAI_API_KEY": "your-openai-api-key-here"
      }
    }
  }
}

For Cline (~/.config/cline/mcp_settings.json):

{
  "mcpServers": {
    "fast-pyairbyte": {
      "command": "npx",
      "args": ["fast-pyairbyte"],
      "env": {
        "OPENAI_API_KEY": "your-openai-api-key-here"
      }
    }
  }
}

Real world lessons from building MCP servers

Quinton — Fri, 20 Jun 2025 19:30:26 +0000

MCP servers are everywhere now. Whether you are using tools like Claude Desktop, ChatGPT, Cursor, Cline, Postman, you name it; If a developer can plug in an MCP to it, they will. Having built a number of MCP servers recently at Airbyte, and running my own side hustle at mycaminoguide.com as an AI agent for the past year, I've learned a few things about what it really takes to build and run MCP servers.

Know the main components of an MCP server.

This might sound obvious, but knowing the main components of an MCP server is incredibly helpful as you build your own. It gives you a roadmap on what and how you want to implement services, and will save you significant time trying to custom code your own solutions when you should have used native MCP decorators and support.

Tools: Tools are functions clients can call to perform an action. Tools are what show up in something like Cursor. When designing an MCP, I typically start by thinking about the domain I want to work in. eg: functions a developer needs when working with my product, or calendar functions etc. From there I have a scope and can decide what actaul functions are available. If I find I am exceeding the domain, I typically create a separate MCP server.

Resources: Resources allow your client to return specific data based on parameters. Resources are very helpful if your MCP service is going to perform some sort of query on a backend system. eg: My MCP offers a calendar service and I want to pass in a particular date to get availability.

@mcp.resource("availability://{user_id}/date")

Prompts: Prompts are messages templates that include parameterized values that you can pass to an LLM to perform a query. I use prompts extensively within the PyAirbyte MCP server to allow the user to specify source and destination connectors. The MCP server then uses a consistent prompt and the OpenAI chat completion API to query the vector store for highly relevant results.

Understand the Transports

Clients support different transports depending on your deployment model.

stdio

If you are running locally, the transport is going to be stdio. Effectively, you configuring your mcp to execute a shell command to run a local file. I use stdio MCP services that I have built to help me automate frequent daily tasks such as checking the health of pipelines, looking at usage analytics, slack summaries etc from within Claude Desktop. I wouldn't recommend stdio for broader developer community facing tools. There is too much local config that the user needs to manage.

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

SSE

Server Sent Events, or SSE, is the original transport for remote MCP servers. MCP servers built using this model require you to run a server such as Express or FastMCP to serve endpoints, both a POST and GET. Remote servers in general are not supported by Claude Desktop, but are supported in Cursor and Cline, although there are limitations, which I'll cover shortly. If you are starting to write MCP servers today, I would not recommend using SSE transports as they have been deprecated in favor of Streamable transports.

Streamable HTTP

Streamable HTTP transports removes the need to create two endpoints - a POST and GET - like you see in SSE transports and are slightly more complex to set up. Once you do have them configured though, there is a lot of benefits through scalability and resumable connections. In addition, they can work stateless meaning you can deploy them quite easily on Vercel vs. SSE services which you need to deploy on something like Railway or Heroku. The downside is that the Streamable HTTP transport is very new with Client tool vendors only now implementing it. There are positive sign though that this transport will become the most dominant. I've already see Claude Code implement a --transport http parameter, for example.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
import asyncio
import json

app = FastAPI()

@app.post("/mcp")
async def handle_mcp(request: Request):
    body = await request.json()

    method = body.get("method")
    if method == "tool/echo":
        params = body.get("params", {})
        stream = params.get("stream", False)

        if stream:
            async def stream_response():
                for i in range(3):
                    event = {
                        "jsonrpc": "2.0",
                        "id": body.get("id"),
                        "result": {"message": f"streaming chunk {i}"}
                    }
                    yield f"data: {json.dumps(event)}\n\n"
                    await asyncio.sleep(0.5)
            return StreamingResponse(stream_response(), media_type="text/event-stream")

        # Non-streaming response
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": body.get("id"),
            "result": {"message": "echo response"}
        })

    return JSONResponse({
        "jsonrpc": "2.0",
        "id": body.get("id"),
        "error": {"code": -32601, "message": "Method not found"}
    })

# Run with: uvicorn app:app --reload

Utilize a good library

Most of my MCP development is done in python. Thankfully, there is a rich ecosystem of libraries available to that make working with MCP much easier.

FastMCP

FastMCP is the defacto standard. It is fully spec-compliant, supports streaming transport, and is easily deployed.

from fastmcp import MCPApp
app = MCPApp()

@app.tool("math/add")
def add(a: int, b: int) -> int:
    return a + b

OpenAI Responses API

It's been interesting to see OpenAI support a competitors 'standard' (Anthrophic were the original authors of the MCP spec). As a heavy user of the Responses API in mycaminoguide.com, I've been excited to see that models can now use MCP servers to perform tasks. Currently the implementation doesn't feel very natural and there it's overly complex but the idea of an agent or model using my MCP server has me watching this space closely. Google is also pushing the same approach with their Agent SDK.

Not all Clients are created equal

When it client tools such as Claude Desktop, Cline, and Cursor, etc, the level of support for the MCP spec, and how this is represented in the mcp.json a user needs to add to connect a server can often lead to wasted time trying to figure out why an error is being raised. I have not found a centralized place where these differences are listed. Here are the ones I have encountered

Local MCP server support: Claude Desktop, Cline, Cursor, Claude Code.
Remote MCP server support: Cline, Cursor, Claude Code
Remote MCP server passing env in mcp.json: Cursor, Claude Code

The remote MCP server with support for passing environment variables is a interesting case. For example, we just deployed an MCP server for PyAirbyte. This server uses openAI and a vector store to generate data pipelines. It is deployed on Heroku. As part of the client config, we require that you pass in your OpenAI API key. This works great within Cursor, but unfortunately it not supported in Cline. You can, of course, add values to a serverside .env file, but we did not want to do this due to the risk of someone spamming the MCP server and running up a bit OpenAI bill.

{
  "mcpServers": {
    "pyairbyte-mcp": {
      "url": "https://pyairbyte-mcp-7b7b8566f2ce.herokuapp.com/mcp",
      "env": {
        "OPENAI_API_KEY": "your-api-key"
      }
    }
  }
}

Summary

MCP protocols are still evolving. Change is constant and can be frustrating when building services. Sometimes logging errors are not very helpful, and LLMs like ChatGPT often send you down a rabbit hole, only to find out that the spec has changed and the LLM doesn't have the most recent information. Vibe coding MCP servers can be an exercise in frustration. I hope these tips help you get started in building your own MCP servers and avoid some of the pitfalls I made when starting out.

Creating Stripe Test Data in Python

Quinton — Thu, 26 Dec 2024 19:31:24 +0000

We've been working on a new AI+data course to show how you can build a AI chatbot by moving e-commerce data from Stripe into PGVector running on Supabase, via Airbyte PGVector connector to create OpenAI embeddings, using OpenAI client libraries to add natural language support into an app. This is a pretty common "intelligent data stack" app pattern that many of our customers are implementing. The source and destination may change, but pattern (data source > move data and create embeddings > vector-enabled data store > web app with OpenAI) stays the same.

Since we are working on course that is intended for folks to get hands on with, we wanted to make set up as easy as possible. A big part of this was creating sufficient test data in Stripe so there would be a reasonable dataset for the chatbot to interact with. If you've used Stripe before, you know they have a great Sandbox where you can experiment with. The only problem is that it doesn't have sample data pre-loaded.

There are a few sample datasets you can load via the CLI fixtures command. But, for our use, these didn't fit the need. We wanted a larger data set, and since this material will be used online and in workshops, asking learners to install something, like the CLI, on their local machines opens you up to a whole bunch of complexity dealing. You never know what OS version the user is running, whether they have the correct permissions to install things, and so much more. I've been burnt too many times to go down that road.

Thankfully, Stripe also has fantastic APIs, and a great Python client which means we could quickly create a collab notebook for learners to run and insert the data we wanted.

After installing the stripe library via !pip install stripe and passing in a test key using Google Collab secrets, we had to set up some random names for customers and products. The goal was to insert a random collection customers, products with different prices, and purchases. This way when we ask the chatbot questions like "who made the cheapest purchase? How much did they pay, and what did they buy?" there was sufficient data.

import stripe
import random
from google.colab import userdata


stripe.api_key = userdata.get('STRIPE_TEST_KEY')


# Sample data for generating random names
first_names = ["Alice", "Bob", "Charlie", "Diana", "Eve", "Frank", "Grace", "Hank", "Ivy", "Jack", "Quinton", "Akriti", "Justin", "Marcos"]
last_names = ["Smith", "Johnson", "Williams", "Jones", "Brown", "Davis", "Miller", "Wilson", "Moore", "Taylor", "Wall", "Chau", "Keswani", "Marx"]
# Sample clothing product names
clothing_names = [
    "T-Shirt", "Jeans", "Jacket", "Sweater", "Hoodie",
    "Shorts", "Dress", "Blouse", "Skirt", "Pants",
    "Shoes", "Sandals", "Sneakers", "Socks", "Hat",
    "Scarf", "Gloves", "Coat", "Belt", "Tie",
    "Tank Top", "Cardigan", "Overalls", "Tracksuit", "Polo Shirt",
    "Cargo Pants", "Capris", "Dungarees", "Boots", "Cufflinks",
    "Raincoat", "Peacoat", "Blazer", "Slippers", "Underwear",
    "Leggings", "Windbreaker", "Tracksuit Bottoms", "Beanie", "Bikini"
]
# List of random colors
colors = [
    "Red", "Blue", "Green", "Yellow", "Black", "White", "Gray",
    "Pink", "Purple", "Orange", "Brown", "Teal", "Navy", "Maroon",
    "Gold", "Silver", "Beige", "Lavender", "Turquoise", "Coral"
]

Next, it was time to add functions for each of the data types in Stripe that we needed.


# Function to create sample customers with random names
def create_customers(count=5):
    customers = []
    for _ in range(count):
        first_name = random.choice(first_names)
        last_name = random.choice(last_names)
        name = f"{first_name} {last_name}"
        email = f"{first_name.lower()}.{last_name.lower()}@example.com"

        customer = stripe.Customer.create(
            name=name,
            email=email,
            description="Sample customer for testing"
        )
        customers.append(customer)
        print(f"Created Customer: {customer['name']} (ID: {customer['id']})")
    return customers

# Function to create sample products with random clothing names and colors
def create_products(count=3):
    products = []
    for _ in range(count):
        color = random.choice(colors)
        product_name = random.choice(clothing_names)
        full_name = f"{color} {product_name}"
        product = stripe.Product.create(
            name=full_name,
            description=f"This is a {color.lower()} {product_name.lower()}"
        )
        products.append(product)
        print(f"Created Product: {product['name']} (ID: {product['id']})")
    return products

# Function to create prices for the products with random unit_amount
def create_prices(products, min_price=500, max_price=5000):
    prices = []
    for product in products:
        unit_amount = random.randint(min_price, max_price)  # Random amount in cents
        price = stripe.Price.create(
            unit_amount=unit_amount,
            currency="usd",
            product=product['id']
        )
        prices.append(price)
        print(f"Created Price: ${unit_amount / 100:.2f} for Product {product['name']} (ID: {price['id']})")
    return prices

# Function to create random purchases for each customer
def create_purchases(customers, prices, max_purchases_per_customer=5):
    purchases = []
    for customer in customers:
        num_purchases = random.randint(1, max_purchases_per_customer)  # Random number of purchases per customer
        for _ in range(num_purchases):
            price = random.choice(prices)  # Randomly select a product's price
            purchase = stripe.PaymentIntent.create(
                amount=price['unit_amount'],  # Amount in cents
                currency=price['currency'],
                customer=customer['id'],
                payment_method_types=["card"],  # Simulate card payment
                description=f"Purchase of {price['product']} by {customer['name']}"
            )
            purchases.append(purchase)
            print(f"Created Purchase for Customer {customer['name']} (Amount: ${price['unit_amount'] / 100:.2f})")
    return purchases

All that was left is to run the script and specify how much data we need.

# Main function to create sample data
def main():
    print("Creating sample customers with random names...")
    customers = create_customers(count=20)

    print("\nCreating sample products with random clothing names and colors...")
    products = create_products(count=30)

    print("\nCreating prices for products with random amounts...")
    prices = create_prices(products, min_price=500, max_price=5000)

    print("\nCreating random purchases for each customer...")
    purchases = create_purchases(customers, prices, max_purchases_per_customer=10)

    print("\nSample data creation complete!")
    print(f"Created {len(customers)} customers, {len(products)} products, and {len(purchases)} purchases.")

if __name__ == "__main__":
    main()

With the data loaded into our Stripe Sandbox, hooking it up to Airbyte only took a few minutes by using the Connector Builder to map the API endpoints to streams for each data type and setting up a sync job.

Problem solved! Our Collab Python script is super easy for the learner to insert test data into Stripe. Hope it's helpful for someone else doing similar testing.

Can AI finally generate best practice code? I think so.

Quinton — Thu, 19 Dec 2024 21:05:14 +0000

The other week I posted about best practices and pro tips on creating prompts with Cursor. At the heart of these tips was the idea of providing the composer as much context as possible.

In Cursor you could this through two primary ways:

By @mentioning docs or code directly into the composer window or via Cursor settings
Generally by creating more prescriptive prompts such as “always use .env files for storing credentials”

Using @mentions in Cursor

If there is anything you get used to working in tech, and especially now with AI, is the rapid pace of change.

Since I wrote that post, a LOT has changed. Tools like Cursor and Windsurf have introduced rules files. Rules files, and understanding which LLM are best for what task, are a game changer for how I’ve been working with prompts and coding.

You can think of rules files are a place where you can specify longer, more prescriptive rules that your prompts in your project must follow. If you are used to working with dependency files in coding, it’s kind of like this but on steroids.

At the same time, we’ve moved from generative AI assistants to collaborative agentic AI.

Put these together, and the results are pretty amazing!

Here’s an example.

Let’s say I’m using Cursor to build a bunch of data apps and using Airbyte as the data movement platform and Streamlit for the frontend. I’m writing in Python and using the Airbyte API libraries. This is my basic ‘tech stack’.

I can define this in my .cursorrules file that I put in the root of my project folder.

- name: "Import Required Libraries"
    action: "Always ensure the required libraries are imported at the top of the file."
    details:
      - "os: To load environment variables."
      - "dotenv: To load .env file for credentials."
      - "pandas: For mapping API results to DataFrame."
      - "airbyte_api_client: For interacting with Airbyte APIs."
      - "airbyte_api_client.api: Import API-specific modules."
      - "airbyte_api_client.configuration: To configure the API client."
      - "requests: For token management and refreshing the token."

I can go further as well and tell Cursor some of my coding preferences, such as how I want to authenticate, store sensitive keys, and how to work with data.

- name: "Store Sensitive Information Securely"
    action: "All API keys, tokens, client IDs, and client secrets must be stored in a .env file."
    details:
      - ".env file example:"
        content: |
          AIRBYTE_API_URL=https://your-airbyte-instance.com
          AIRBYTE_CLIENT_ID=your_client_id
          AIRBYTE_CLIENT_SECRET=your_client_secret
      - "Load credentials using the python-dotenv library."
      - "Example code:"
        content: |
          from dotenv import load_dotenv
          import os
          load_dotenv()
          AIRBYTE_API_URL = os.getenv('AIRBYTE_API_URL')
          AIRBYTE_CLIENT_ID = os.getenv('AIRBYTE_CLIENT_ID')
          AIRBYTE_CLIENT_SECRET = os.getenv('AIRBYTE_CLIENT_SECRET')

  - name: "Authenticate Using Bearer Token"
    action: "Use client and secret keys to fetch a new bearer token before making API requests."
    details:
      - "Add token refresh logic in a reusable function."
      - "Example function to get a fresh token:"
        content: |
          import requests

          def get_access_token():
              """
              Fetch a new access token using client ID and secret.

              Returns:
                  str: Bearer token.
              """
              url = f"{AIRBYTE_API_URL}/oauth/token"
              payload = {
                  "client_id": AIRBYTE_CLIENT_ID,
                  "client_secret": AIRBYTE_CLIENT_SECRET,
                  "grant_type": "client_credentials"
              }
              response = requests.post(url, json=payload)
              response.raise_for_status()
              return response.json()["access_token"]

Here’s where things really get interesting. I want my code to be easy to maintain and well structured. Early genAI assistants really struggled here. They often generated monolithic blocks of code, that whilst they worked, they were pretty ugly. I want my rules to tell Cursor how I like my code. Thankfully, agentic AI can use reasoning to figure out how to create well structured code. In the world of Python, PEP 8 provides a style guide for best practices on how to write good code. I can now tell Cursor to use this guide whenever it writes code!

- name: "Follow PEP8 Naming Conventions"
    action: "Ensure that variable, function, and class names adhere to PEP8 standards."
    details:
      - "Use snake_case for variable and function names."
      - "Use PascalCase for class names."
      - "Limit line length to 79 characters."
      - "Prefix all private variables and functions with an underscore."
       - "Use encapsulation for best practices"

Now, when I create rules, or generate code, my functions are neat, easy to read, and practices good encapsulation technics. All the things I would -- well, should -- do when coding.

 - name: "Map Results to pandas DataFrame"
    action: "Always convert API response data to pandas DataFrame for further processing."
    details:
      - "Example function to convert API response to DataFrame:"
        content: |
          import pandas as pd

          def map_to_dataframe(data, columns=None):
              """
              Convert API response data to pandas DataFrame.

              Args:
                  data (list or dict): JSON response data.
                  columns (list, optional): List of column names to include.

              Returns:
                  pd.DataFrame: Data in tabular format.
              """
              df = pd.DataFrame(data)
              if columns:
                  df = df[columns]
              return df

PEP 8, also offers guidelines for documenting code, how to handle error messages, and much more. I can add examples in my .cursorrules file that problems can use as context whenever creating new code.

- name: "Error Handling for API Requests"
    action: "Implement robust error handling for API calls."
    details:
      - "Use try-except blocks to handle common exceptions."
      - "Raise errors for debugging when necessary."
      - "Example:"
        content: |
          try:
              api_client = get_api_client()
              df = get_workspaces(api_client)
              print(df.head())
          except Exception as e:
              print(f"An error occurred: {e}")

Putting it all together, my current .cursorrules file for Airbyte development in Python means I can spend more time thinking about the task I want the prompt to perform vs. how to structure the actual prompt to create best practice code. This has saved me hours and hours of refactoring time.

And, since I switch between Python, React, and Swift on a regular basis, I’ve started creating rules files for each language. This approach let’s me apply best practices per language, define things like directory structures in my projects and much more. If you want to check out my rules files, I’ve added them to a GitHub repo. I'm constantly refactoring and adding more. Star and follow along if you want to stay up to date.

# CursorRules file for Xcode Swift Project
# Defines best practices for directory structure

# Base directory
ProjectRoot:
  - README.md           # Project documentation
  - LICENSE             # License file
  - .gitignore          # Git ignore rules
  - .xcodeproj          # Xcode project file
  - .xcworkspace        # Xcode workspace file (if using CocoaPods or similar)

# Primary folders
Sources:                # All source code files
  - App:                # Main app folder
    - Models:           # Data models
    - Views:            # User interface components
    - ViewModels:       # Logic connecting views and models (MVVM pattern)
    - Controllers:      # View controllers (MVC pattern)
  - Resources:          # Assets and other resources
    - Assets.xcassets   # Image and color assets
    - LaunchScreen.storyboard # Launch screen storyboard
    - Localizations:    # Localized strings and assets

Tests:                  # Unit and UI tests
  - UnitTests:          # Unit test cases
  - UITests:            # UI test cases

# Supporting files
Configurations:         # Build and app configurations
  - Debug.xcconfig      # Debug configuration settings
  - Release.xcconfig    # Release configuration settings

Extensions:             # Code extensions (e.g., for UIKit or SwiftUI)
  - String+Extensions.swift
  - Date+Extensions.swift

Utilities:              # General utilities and helpers
  - NetworkManager.swift
  - APIConstants.swift

# Example
Example:                # Example apps or demos
  - ExampleApp:         # Example implementation using the project

Documentation:          # Developer documentation
  - Docs.md             # Project documentation
  - Tutorials:          # Tutorials for using the app

# Notes
- Organize files logically into respective folders based on MVC or MVVM architecture.
- Ensure all resources are in `Resources`.
- Follow naming conventions and avoid file duplication.
- Unit tests mirror the structure of the `Sources` folder.

But, back to my Airbyte + Streamlit app. Now that I have my rules file created, all I need to do is write a simple prompt and let AI do the work

create a streamlit app that connects to my airbyte cloud workspace and 
lists available connections, and let's me click on them to show the stream available.

As you can see, the code gets generated, and composer gives me a run down of the rules that it has followed to create it. I honestly don't know how I ever lived without it. Rules files are a game changer for developers and prompts.