# Directory Structure
```
├── .gitignore
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── pytest.ini
├── README.md
├── smithery.yaml
├── src
│ └── flights
│ ├── __init__.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── client.py
│ │ └── endpoints.py
│ ├── config
│ │ ├── __init__.py
│ │ └── api.py
│ ├── models
│ │ ├── flight_search.py
│ │ ├── multi_city.py
│ │ ├── offers.py
│ │ ├── search.py
│ │ ├── segments.py
│ │ └── time_specs.py
│ ├── server.py
│ └── services
│ ├── __init__.py
│ └── search.py
├── tests
│ ├── __init__.py
│ └── test_duffel_api.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
.env
.venv
env/
venv/
ENV/
# IDE
.idea/
.vscode/
*.swp
*.swo
.DS_Store
# Testing
.coverage
htmlcov/
.pytest_cache/
.tox/
# Logs
*.log
# Local development
.python-version
.env.local
.env.*.local
# Distribution
dist/
build/
# UV
.uv/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Find Flights MCP Server
MCP server for searching and retrieving flight information using Duffel API.
## How it Works
## Video Demo
https://github.com/user-attachments/assets/c111aa4c-9559-4d74-a2f6-60e322c273d4
## Why This is Helpful
While tools like Google Flights work great for simple trips, this tool shines when dealing with complex travel plans. Here's why:
- **Contextual Memory**: Claude remembers all your previous flight searches in the chat, so you don't need to keep multiple tabs open to compare prices
- **Flexible Date Search**: Easily search across multiple days to find the best prices without manually checking each date
- **Complex Itineraries**: Perfect for multi-city trips, one-stop flights, or when you need to compare different route options you can just ask!
- **Natural Conversation**: Just describe what you're looking for - no more clicking through calendar interfaces or juggling search parameters down to parsing city names, dates, and times.
Think of it as having a travel agent in your chat who remembers everything you've discussed and can instantly search across dates and routes.
## Features
- Search for flights between multiple destinations
- Support for one-way, round-trip, and multi-city flight queries
- Detailed flight offer information
- Flexible search parameters (departure times, cabin class, number of passengers)
- Automatic handling of flight connections
- Search for flights within multiple days to find the best flight for your trip (slower)
## Prerequisites
- Python 3.x
- Duffel API Live Key
## Getting Your Duffel API Key
Duffel requires account verification and payment information setup, but this MCP server only uses the API for searching flights - no actual bookings or charges will be made to your account.
Try using duffel_test first to see the power of this tool. If you end up liking it, you can go through the verification process below to use the live key.
### Test Mode First (Recommended)
You can start with a test API key (`duffel_test`) to try out the functionality with simulated data before going through the full verification process:
1. Visit [Duffel's registration page](https://app.duffel.com/join)
2. Create an account (you can select "Personal Use" for Company Name)
3. Navigate to More > Developer to find your test API key (one is already provided)
### Getting a Live API Key
To access real flight data, follow these steps:
1. In the Duffel dashboard, toggle "Test Mode" off in the top left corner
2. The verification process requires multiple steps - you'll need to toggle test mode off repeatedly:
- First toggle: Verify your email address
- Toggle again: Complete company information (Personal Use is fine)
- Toggle again: Add payment information (required by Duffel but NO CHARGES will be made by this MCP server)
- Toggle again: Complete any remaining verification steps
- Final toggle: Access live mode after clicking "Agree and Submit"
3. Once fully verified, go to More > Developer > Create Live Token
4. Copy your live API key
💡 TIP: Each time you complete a verification step, you'll need to toggle test mode off again to proceed to the next step. Keep toggling until you've completed all requirements.
⚠️ IMPORTANT NOTES:
- Your payment information is handled directly by Duffel and is not accessed or stored by the MCP server
- This MCP server is READ-ONLY - it can only search for flights, not book them
- No charges will be made to your payment method through this integration
- All sensitive information (including API keys) stays local to your machine
- You can start with the test API key (`duffel_test`) to evaluate the functionality
- The verification process may take some time - this is a standard Duffel requirement
### Security Note
This MCP server only uses Duffel's search endpoints and cannot make bookings or charges. Your payment information is solely for Duffel's verification process and is never accessed by or shared with the MCP server.
### Note on API Usage Limits
- Check Duffel's current pricing and usage limits
- Different tiers available based on your requirements
- Recommended to review current pricing on their website
## Installation
### Installing via Smithery
To install Find Flights for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@ravinahp/travel-mcp):
```bash
npx -y @smithery/cli install @ravinahp/travel-mcp --client claude
```
### Manual Installation
Clone the repository:
```bash
git clone https://github.com/ravinahp/flights-mcp
cd flights-mcp
```
Install dependencies using uv:
```bash
uv sync
```
Note: We use uv instead of pip since the project uses pyproject.toml for dependency management.
## Configure as MCP Server
To add this tool as an MCP server, modify your Claude desktop configuration file.
Configuration file locations:
- MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
- Windows: `%APPDATA%/Claude/claude_desktop_config.json`
Add the following configuration to your JSON file:
```json
{
"flights-mcp": {
"command": "uv",
"args": [
"--directory",
"/Users/YOUR_USERNAME/Code/flights-mcp",
"run",
"flights-mcp"
],
"env": {
"DUFFEL_API_KEY_LIVE": "your_duffel_live_api_key_here"
}
}
}
```
⚠️ IMPORTANT:
- Replace `YOUR_USERNAME` with your actual system username
- Replace `your_duffel_live_api_key_here` with your actual Duffel Live API key
- Ensure the directory path matches your local installation
## Deployment
### Building
Prepare the package:
```bash
# Sync dependencies and update lockfile
uv sync
# Build package
uv build
```
This will create distributions in the `dist/` directory.
## Debugging
For the best debugging experience, use the MCP Inspector:
```bash
npx @modelcontextprotocol/inspector uv --directory /path/to/find-flights-mcp run flights-mcp
```
The Inspector provides:
- Real-time request/response monitoring
- Input/output validation
- Error tracking
- Performance metrics
## Available Tools
### 1. Search Flights
```python
@mcp.tool()
async def search_flights(params: FlightSearch) -> str:
"""Search for flights based on parameters."""
```
Supports three flight types:
- One-way flights
- Round-trip flights
- Multi-city flights
Parameters include:
- `type`: Flight type ('one_way', 'round_trip', 'multi_city')
- `origin`: Origin airport code
- `destination`: Destination airport code
- `departure_date`: Departure date (YYYY-MM-DD)
- Optional parameters:
- `return_date`: Return date for round-trips
- `adults`: Number of adult passengers
- `cabin_class`: Preferred cabin class
- `departure_time`: Specific departure time range
- `arrival_time`: Specific arrival time range
- `max_connections`: Maximum number of connections
### 2. Get Offer Details
```python
@mcp.tool()
async def get_offer_details(params: OfferDetails) -> str:
"""Get detailed information about a specific flight offer."""
```
Retrieves comprehensive details for a specific flight offer using its unique ID.
### 3. Search Multi-City Flights
```python
@mcp.tool(name="search_multi_city")
async def search_multi_city(params: MultiCityRequest) -> str:
"""Search for multi-city flights."""
```
Specialized tool for complex multi-city flight itineraries.
Parameters include:
- `segments`: List of flight segments
- `adults`: Number of adult passengers
- `cabin_class`: Preferred cabin class
- `max_connections`: Maximum number of connections
## Use Cases
### Some Example (But try it out yourself!)
You can use these tools to find flights with various complexities:
- "Find a one-way flight from SFO to NYC on Jan 7 for 2 adults in business class"
- "Search for a round-trip flight from LAX to London, departing Jan 8 and returning Jan 15"
- "Plan a multi-city trip from New York to Paris on Jan 7, then to Rome on Jan 10, and back to New York on Jan 15"
- "What is the cheapest flight from SFO to LAX from Jan 7 to Jan 15 for 2 adults in economy class?"
- You can even search for flights within multiple days to find the best flight for your trip. Right now, the reccomendation is to only search for one-way or round-trip flights this way. Example: "Find the cheapest flight from SFO to LAX from Jan 7 to Jan 10 for 2 adults in economy class"
## Response Format
The tools return JSON-formatted responses with:
- Flight offer details
- Pricing information
- Slice (route) details
- Carrier information
- Connection details
## Error Handling
The service includes robust error handling for:
- API request failures
- Invalid airport codes
- Missing or invalid API keys
- Network timeouts
- Invalid search parameters
## Contributing
[Add guidelines for contribution, if applicable]
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Performance Notes
- Searches are limited to 50 offers for one-way/round-trip flights
- Multi-city searches are limited to 10 offers
- Supplier timeout is set to 15-30 seconds depending on the search type
### Cabin Classes
Available cabin classes:
- `economy`: Standard economy class
- `premium_economy`: Premium economy class
- `business`: Business class
- `first`: First class
Example request with cabin class:
```json
{
"params": {
"type": "one_way",
"adults": 1,
"origin": "SFO",
"destination": "LAX",
"departure_date": "2025-01-12",
"cabin_class": "business" // Specify desired cabin class
}
}
```
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
# Tests package initialization
```
--------------------------------------------------------------------------------
/src/flights/api/__init__.py:
--------------------------------------------------------------------------------
```python
"""Duffel API client package."""
from .client import DuffelClient
__all__ = ['DuffelClient']
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
pythonpath = src
log_cli = true
log_cli_level = INFO
```
--------------------------------------------------------------------------------
/src/flights/config/__init__.py:
--------------------------------------------------------------------------------
```python
"""Configuration package."""
from .api import DUFFEL_API_URL, DUFFEL_API_VERSION, get_api_token
__all__ = ['DUFFEL_API_URL', 'DUFFEL_API_VERSION', 'get_api_token']
```
--------------------------------------------------------------------------------
/src/flights/services/__init__.py:
--------------------------------------------------------------------------------
```python
"""Flight search services."""
from .search import search_flights, get_offer_details, search_multi_city
__all__ = ['search_flights', 'get_offer_details', 'search_multi_city']
```
--------------------------------------------------------------------------------
/src/flights/__init__.py:
--------------------------------------------------------------------------------
```python
"""Flight search MCP package initialization."""
from . import server
import asyncio
def main():
"""Main entry point for the package."""
asyncio.run(server.main())
__all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/src/flights/models/offers.py:
--------------------------------------------------------------------------------
```python
"""Offer-related models."""
from pydantic import BaseModel, Field
class OfferDetails(BaseModel):
"""Model for getting detailed offer information."""
offer_id: str = Field(..., description="The ID of the offer to get details for")
```
--------------------------------------------------------------------------------
/src/flights/models/flight_search.py:
--------------------------------------------------------------------------------
```python
"""Flight search models."""
from .search import FlightSearch
from .multi_city import MultiCityRequest
from .segments import FlightSegment
from .offers import OfferDetails
__all__ = [
'FlightSearch',
'MultiCityRequest',
'FlightSegment',
'OfferDetails',
]
```
--------------------------------------------------------------------------------
/src/flights/models/time_specs.py:
--------------------------------------------------------------------------------
```python
"""Time specification models."""
from pydantic import BaseModel, Field
class TimeSpec(BaseModel):
"""Model for time range specification."""
from_time: str = Field(..., description="Start time (HH:MM)", pattern="^([0-1]?[0-9]|2[0-3]):[0-5][0-9]$")
to_time: str = Field(..., description="End time (HH:MM)", pattern="^([0-1]?[0-9]|2[0-3]):[0-5][0-9]$")
```
--------------------------------------------------------------------------------
/src/flights/models/segments.py:
--------------------------------------------------------------------------------
```python
"""Flight segment models."""
from pydantic import BaseModel, Field
class FlightSegment(BaseModel):
"""Model for a single flight segment in a multi-city trip."""
origin: str = Field(..., description="Origin airport code")
destination: str = Field(..., description="Destination airport code")
departure_date: str = Field(..., description="Departure date (YYYY-MM-DD)")
```
--------------------------------------------------------------------------------
/src/flights/config/api.py:
--------------------------------------------------------------------------------
```python
"""Duffel API configuration."""
import os
from typing import Final
# API Constants
DUFFEL_API_URL: Final = "https://api.duffel.com"
DUFFEL_API_VERSION: Final = "v2"
def get_api_token() -> str:
"""Get Duffel API token from environment."""
token = os.getenv("DUFFEL_API_KEY_LIVE")
if not token:
raise ValueError("DUFFEL_API_KEY_LIVE environment variable not set")
return token
```
--------------------------------------------------------------------------------
/src/flights/server.py:
--------------------------------------------------------------------------------
```python
"""Server initialization for find-flights MCP."""
import logging
from .services.search import mcp
# Set up logging
logger = logging.getLogger(__name__)
def main():
"""Entry point for the find-flights-mcp application."""
logger.info("Starting Find Flights MCP server")
try:
mcp.run(transport='stdio')
logger.info("Server initialized successfully")
except Exception as e:
logger.error(f"Server error occurred: {str(e)}", exc_info=True)
raise
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "flights-mcp"
version = "0.1.0"
description = "Flight search MCP server using Duffel API"
requires-python = ">=3.10"
dependencies = [
"httpx",
"python-dotenv",
"pydantic",
"mcp",
]
license = "MIT"
[project.scripts]
flights-mcp = "flights:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/flights"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
python_files = ["test_*.py"]
pythonpath = ["src"]
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/deployments
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- duffelApiKeyLive
properties:
duffelApiKeyLive:
type: string
description: The live API key for accessing the Duffel flight search service.
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({ command: 'flights-mcp', env: { DUFFEL_API_KEY_LIVE: config.duffelApiKeyLive } })
```
--------------------------------------------------------------------------------
/src/flights/models/multi_city.py:
--------------------------------------------------------------------------------
```python
"""Multi-city flight search models."""
from typing import Optional, List, Literal
from pydantic import BaseModel, Field
from .time_specs import TimeSpec
from .segments import FlightSegment
class MultiCityRequest(BaseModel):
"""Model for multi-city flight search."""
type: Literal["multi_city"]
segments: List[FlightSegment] = Field(..., min_items=2, description="Flight segments")
cabin_class: str = Field("economy", description="Cabin class")
adults: int = Field(1, description="Number of adult passengers")
max_connections: int = Field(None, description="Maximum number of connections (0 for non-stop)")
departure_time: TimeSpec | None = Field(None, description="Optional departure time range")
arrival_time: TimeSpec | None = Field(None, description="Optional arrival time range")
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use a Python image with uv pre-installed
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
# Install the project into /app
WORKDIR /app
# Enable bytecode compilation
ENV UV_COMPILE_BYTECODE=1
# Copy from the cache instead of linking since it's a mounted volume
ENV UV_LINK_MODE=copy
# Install the project's dependencies using the lockfile and settings
COPY pyproject.toml uv.lock /app/
RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-install-project --no-dev --no-editable
# Then, add the rest of the project source code and install it
# Installing separately from its dependencies allows optimal layer caching
ADD src /app/src
RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev --no-editable
FROM python:3.12-slim-bookworm
WORKDIR /app
COPY --from=uv /root/.local /root/.local
COPY --from=uv --chown=app:app /app/.venv /app/.venv
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
# Define environment variable for Duffel API key
ENV DUFFEL_API_KEY_LIVE=your_duffel_live_api_key_here
# Start the MCP server
ENTRYPOINT ["flights-mcp"]
```
--------------------------------------------------------------------------------
/src/flights/models/search.py:
--------------------------------------------------------------------------------
```python
"""Flight search models."""
from typing import Optional, List
from pydantic import BaseModel, Field
from .time_specs import TimeSpec
class FlightSearch(BaseModel):
"""Model for flight search parameters."""
type: str = Field(..., description="Type of flight: 'one_way', 'round_trip', or 'multi_city'")
origin: str = Field(..., description="Origin airport code")
destination: str = Field(..., description="Destination airport code")
departure_date: str = Field(..., description="Departure date (YYYY-MM-DD)")
return_date: str | None = Field(None, description="Return date for round trips (YYYY-MM-DD)")
departure_time: TimeSpec | None = Field(None, description="Preferred departure time range")
arrival_time: TimeSpec | None = Field(None, description="Preferred arrival time range")
cabin_class: str = Field("economy", description="Cabin class (economy, business, first)")
adults: int = Field(1, description="Number of adult passengers")
max_connections: int = Field(None, description="Maximum number of connections (0 for non-stop)")
additional_stops: Optional[List[dict]] = Field(None, description="Additional stops for multi-city trips")
```
--------------------------------------------------------------------------------
/src/flights/api/client.py:
--------------------------------------------------------------------------------
```python
"""Duffel API client."""
import logging
import httpx
from typing import Dict, Any, List
from ..config import get_api_token
from .endpoints import OfferEndpoints
class DuffelClient:
"""Client for interacting with the Duffel API."""
def __init__(self, logger: logging.Logger, timeout: float = 30.0):
"""Initialize the Duffel API client."""
self.logger = logger
self.timeout = timeout
self._token = get_api_token()
self.base_url = "https://api.duffel.com/air"
# Headers setup
self.headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
"Duffel-Version": "v2",
"Authorization": f"Bearer {self._token}",
"Content-Type": "application/json"
}
self.logger.info(f"API key starts with: {self._token[:8] if self._token else None}")
self.logger.info(f"Using base URL: {self.base_url}")
# Initialize endpoints
self.offers = OfferEndpoints(self.base_url, self.headers, self.logger)
async def __aenter__(self):
"""Async context manager entry."""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit."""
pass
async def create_offer_request(self, **kwargs) -> Dict[str, Any]:
"""Create an offer request."""
return await self.offers.create_offer_request(**kwargs)
async def get_offer(self, offer_id: str) -> Dict[str, Any]:
"""Get offer details."""
return await self.offers.get_offer(offer_id)
```
--------------------------------------------------------------------------------
/src/flights/api/endpoints.py:
--------------------------------------------------------------------------------
```python
"""Duffel API endpoint handlers."""
from typing import Dict, Any, List
import logging
import httpx
class OfferEndpoints:
"""Offer-related API endpoints."""
def __init__(self, base_url: str, headers: Dict, logger: logging.Logger):
self.base_url = base_url
self.headers = headers
self.logger = logger
async def create_offer_request(
self,
slices: List[Dict],
cabin_class: str = "economy",
adult_count: int = 1,
max_connections: int = None,
return_offers: bool = True,
supplier_timeout: int = 15000
) -> Dict:
"""Create a flight offer request."""
try:
# Format request data
request_data = {
"data": {
"slices": slices,
"passengers": [{"type": "adult"} for _ in range(adult_count)],
"cabin_class": cabin_class,
}
}
if max_connections is not None:
request_data["data"]["max_connections"] = max_connections
params = {
"return_offers": str(return_offers).lower(),
"supplier_timeout": supplier_timeout
}
async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
self.logger.info(f"Creating offer request with data: {request_data}")
response = await client.post(
f"{self.base_url}/offer_requests",
headers=self.headers,
params=params,
json=request_data
)
response.raise_for_status()
data = response.json()
request_id = data["data"]["id"]
offers = data["data"].get("offers", [])
self.logger.info(f"Created offer request with ID: {request_id}")
self.logger.info(f"Received {len(offers)} offers")
return {
"request_id": request_id,
"offers": offers
}
except Exception as e:
error_msg = f"Error creating offer request: {str(e)}"
self.logger.error(error_msg)
raise
async def get_offer(self, offer_id: str) -> Dict:
"""Get details of a specific offer."""
try:
if not offer_id.startswith("off_"):
raise ValueError("Invalid offer ID format - must start with 'off_'")
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/offers/{offer_id}",
headers=self.headers
)
response.raise_for_status()
return response.json()
except Exception as e:
self.logger.error(f"Error getting offer {offer_id}: {str(e)}")
raise
```
--------------------------------------------------------------------------------
/tests/test_duffel_api.py:
--------------------------------------------------------------------------------
```python
"""Tests for Duffel API client."""
import pytest
import logging
from datetime import datetime, timedelta
from flights.api import DuffelClient
from flights.models.search import FlightSearch
from flights.models.multi_city import MultiCityRequest
# Setup logging for tests
logger = logging.getLogger(__name__)
@pytest.fixture
async def client():
"""Create a test client."""
client = DuffelClient(logger)
async with client as c:
yield c
@pytest.mark.asyncio
async def test_search_one_way(client):
"""Test one-way flight search."""
# Get tomorrow's date for testing
tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
response = await client.create_offer_request(
slices=[{
"origin": "SFO",
"destination": "LAX",
"departure_date": tomorrow
}],
cabin_class="economy",
adult_count=1
)
assert response is not None
assert "request_id" in response
assert "offers" in response
assert len(response["offers"]) > 0
@pytest.mark.asyncio
async def test_search_round_trip(client):
"""Test round-trip flight search."""
# Get dates for testing
departure = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")
return_date = (datetime.now() + timedelta(days=14)).strftime("%Y-%m-%d")
response = await client.create_offer_request(
slices=[
{
"origin": "SFO",
"destination": "LAX",
"departure_date": departure
},
{
"origin": "LAX",
"destination": "SFO",
"departure_date": return_date
}
],
cabin_class="economy",
adult_count=1
)
assert response is not None
assert "request_id" in response
assert "offers" in response
assert len(response["offers"]) > 0
@pytest.mark.asyncio
async def test_search_multi_city(client):
"""Test multi-city flight search."""
# Get dates for testing
first_date = (datetime.now() + timedelta(days=10)).strftime("%Y-%m-%d")
second_date = (datetime.now() + timedelta(days=15)).strftime("%Y-%m-%d")
response = await client.create_offer_request(
slices=[
{
"origin": "SFO",
"destination": "LAX",
"departure_date": first_date
},
{
"origin": "LAX",
"destination": "JFK",
"departure_date": second_date
}
],
cabin_class="economy",
adult_count=1
)
assert response is not None
assert "request_id" in response
assert "offers" in response
@pytest.mark.asyncio
async def test_cabin_classes(client):
"""Test different cabin classes."""
tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
cabin_classes = ["economy", "premium_economy", "business", "first"]
for cabin_class in cabin_classes:
response = await client.create_offer_request(
slices=[{
"origin": "SFO",
"destination": "LAX",
"departure_date": tomorrow
}],
cabin_class=cabin_class,
adult_count=1
)
assert response is not None
assert "request_id" in response
assert "offers" in response
@pytest.mark.asyncio
async def test_get_offer(client):
"""Test getting offer details."""
# First create an offer request
tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
offers_response = await client.create_offer_request(
slices=[{
"origin": "SFO",
"destination": "LAX",
"departure_date": tomorrow
}],
cabin_class="economy",
adult_count=1
)
assert offers_response is not None
assert "offers" in offers_response
assert len(offers_response["offers"]) > 0
# Get the first offer's details
offer_id = offers_response["offers"][0]["id"]
offer_details = await client.get_offer(offer_id)
assert offer_details is not None
assert "data" in offer_details
@pytest.mark.asyncio
async def test_error_handling(client):
"""Test error handling for invalid requests."""
with pytest.raises(Exception):
await client.create_offer_request(
slices=[{
"origin": "INVALID",
"destination": "ALSO_INVALID",
"departure_date": "2025-01-01"
}],
cabin_class="economy",
adult_count=1
)
@pytest.mark.asyncio
async def test_invalid_offer_id(client):
"""Test error handling for invalid offer ID."""
with pytest.raises(ValueError):
await client.get_offer("invalid_offer_id")
```
--------------------------------------------------------------------------------
/src/flights/services/search.py:
--------------------------------------------------------------------------------
```python
"""Flight search tools using Duffel API."""
import logging
from typing import Dict
import json
from mcp.server.fastmcp import FastMCP
# Import all models through flight_search
from ..models.flight_search import (
FlightSearch,
MultiCityRequest,
OfferDetails
)
from ..models.time_specs import TimeSpec
from ..api import DuffelClient
# Set up logging
logger = logging.getLogger(__name__)
# Initialize FastMCP server and API client
mcp = FastMCP("find-flights-mcp")
flight_client = DuffelClient(logger)
def _create_slice(origin: str, destination: str, date: str,
departure_time: TimeSpec | None = None,
arrival_time: TimeSpec | None = None) -> Dict:
"""Helper to create a slice with time ranges."""
slice_data = {
"origin": origin,
"destination": destination,
"departure_date": date,
"departure_time": {
"from": "00:00",
"to": "23:59"
},
"arrival_time": {
"from": "00:00",
"to": "23:59"
}
}
if departure_time:
slice_data["departure_time"] = {
"from": departure_time.from_time,
"to": departure_time.to_time
}
if arrival_time:
slice_data["arrival_time"] = {
"from": arrival_time.from_time,
"to": arrival_time.to_time
}
return slice_data
@mcp.tool()
async def search_flights(params: FlightSearch) -> str:
"""Search for flights based on parameters."""
try:
slices = []
# Build slices based on flight type
if params.type == "one_way":
slices = [_create_slice(
params.origin,
params.destination,
params.departure_date,
params.departure_time,
params.arrival_time
)]
elif params.type == "round_trip":
if not params.return_date:
raise ValueError("Return date required for round-trip flights")
slices = [
_create_slice(
params.origin,
params.destination,
params.departure_date,
params.departure_time,
params.arrival_time
),
_create_slice(
params.destination,
params.origin,
params.return_date,
params.departure_time,
params.arrival_time
)
]
elif params.type == "multi_city":
if not params.additional_stops:
raise ValueError("Additional stops required for multi-city flights")
# First leg
slices.append({
"origin": params.origin,
"destination": params.destination,
"departure_date": params.departure_date,
"departure_time": {
"from": "00:00",
"to": "23:59"
},
"arrival_time": {
"from": "00:00",
"to": "23:59"
}
})
# Additional legs
for stop in params.additional_stops:
slices.append({
"origin": stop["origin"],
"destination": stop["destination"],
"departure_date": stop["departure_date"],
"departure_time": {
"from": "00:00",
"to": "23:59"
},
"arrival_time": {
"from": "00:00",
"to": "23:59"
}
})
# Use async context manager
async with flight_client as client:
response = await client.create_offer_request(
slices=slices,
cabin_class=params.cabin_class,
adult_count=params.adults,
max_connections=params.max_connections,
return_offers=True,
supplier_timeout=15000
)
# Format the response
formatted_response = {
'request_id': response['request_id'],
'offers': []
}
# Get all offers (limit to 10 to manage response size)
for offer in response.get('offers', [])[:50]: # Keep the slice to limit offers
offer_details = {
'offer_id': offer.get('id'),
'price': {
'amount': offer.get('total_amount'),
'currency': offer.get('total_currency')
},
'slices': []
}
# Only include essential slice details
for slice in offer.get('slices', []):
segments = slice.get('segments', [])
if segments: # Check if there are any segments
slice_details = {
'origin': slice['origin']['iata_code'],
'destination': slice['destination']['iata_code'],
'departure': segments[0].get('departing_at'), # First segment departure
'arrival': segments[-1].get('arriving_at'), # Last segment arrival
'duration': slice.get('duration'),
'carrier': segments[0].get('marketing_carrier', {}).get('name'),
'stops': len(segments) - 1,
'stops_description': 'Non-stop' if len(segments) == 1 else f'{len(segments) - 1} stop{"s" if len(segments) - 1 > 1 else ""}',
'connections': []
}
# Add connection information if there are multiple segments
if len(segments) > 1:
for i in range(len(segments)-1):
connection = {
'airport': segments[i].get('destination', {}).get('iata_code'),
'arrival': segments[i].get('arriving_at'),
'departure': segments[i+1].get('departing_at'),
'duration': segments[i+1].get('duration')
}
slice_details['connections'].append(connection)
offer_details['slices'].append(slice_details)
formatted_response['offers'].append(offer_details)
return json.dumps(formatted_response, indent=2)
except Exception as e:
logger.error(f"Error searching flights: {str(e)}", exc_info=True)
raise
@mcp.tool()
async def get_offer_details(params: OfferDetails) -> str:
"""Get detailed information about a specific flight offer."""
try:
async with flight_client as client:
response = await client.get_offer(
offer_id=params.offer_id
)
return json.dumps(response, indent=2)
except Exception as e:
logger.error(f"Error getting offer details: {str(e)}", exc_info=True)
raise
@mcp.tool(name="search_multi_city")
async def search_multi_city(params: MultiCityRequest) -> str:
"""Search for multi-city flights."""
try:
slices = []
for segment in params.segments:
slices.append(_create_slice(
segment.origin,
segment.destination,
segment.departure_date,
None,
None
))
# Use async context manager with shorter timeout
async with flight_client as client:
response = await client.create_offer_request(
slices=slices,
cabin_class=params.cabin_class,
adult_count=params.adults,
max_connections=params.max_connections,
return_offers=True,
supplier_timeout=30000 # Increased timeout for multi-city
)
# Format response inside the context
formatted_response = {
'request_id': response['request_id'],
'offers': []
}
# Process offers inside the context
for offer in response.get('offers', [])[:10]:
offer_details = {
'offer_id': offer.get('id'),
'price': {
'amount': offer.get('total_amount'),
'currency': offer.get('total_currency')
},
'slices': []
}
for slice in offer.get('slices', []):
segments = slice.get('segments', [])
if segments:
slice_details = {
'origin': slice['origin']['iata_code'],
'destination': slice['destination']['iata_code'],
'departure': segments[0].get('departing_at'),
'arrival': segments[-1].get('arriving_at'),
'duration': slice.get('duration'),
'carrier': segments[0].get('marketing_carrier', {}).get('name'),
'stops': len(segments) - 1,
'stops_description': 'Non-stop' if len(segments) == 1 else f'{len(segments) - 1} stop{"s" if len(segments) - 1 > 1 else ""}',
'connections': []
}
if len(segments) > 1:
for i in range(len(segments)-1):
connection = {
'airport': segments[i].get('destination', {}).get('iata_code'),
'arrival': segments[i].get('arriving_at'),
'departure': segments[i+1].get('departing_at'),
'duration': segments[i+1].get('duration')
}
slice_details['connections'].append(connection)
offer_details['slices'].append(slice_details)
formatted_response['offers'].append(offer_details)
return json.dumps(formatted_response, indent=2)
except Exception as e:
logger.error(f"Error searching flights: {str(e)}", exc_info=True)
raise
```