# 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:
--------------------------------------------------------------------------------
```
1 | # Python
2 | __pycache__/
3 | *.py[cod]
4 | *$py.class
5 | *.so
6 | .Python
7 | build/
8 | develop-eggs/
9 | dist/
10 | downloads/
11 | eggs/
12 | .eggs/
13 | lib/
14 | lib64/
15 | parts/
16 | sdist/
17 | var/
18 | wheels/
19 | *.egg-info/
20 | .installed.cfg
21 | *.egg
22 |
23 | # Virtual Environment
24 | .env
25 | .venv
26 | env/
27 | venv/
28 | ENV/
29 |
30 | # IDE
31 | .idea/
32 | .vscode/
33 | *.swp
34 | *.swo
35 | .DS_Store
36 |
37 | # Testing
38 | .coverage
39 | htmlcov/
40 | .pytest_cache/
41 | .tox/
42 |
43 | # Logs
44 | *.log
45 |
46 | # Local development
47 | .python-version
48 | .env.local
49 | .env.*.local
50 |
51 | # Distribution
52 | dist/
53 | build/
54 |
55 | # UV
56 | .uv/
57 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Find Flights MCP Server
2 | MCP server for searching and retrieving flight information using Duffel API.
3 |
4 | ## How it Works
5 | 
6 |
7 | ## Video Demo
8 | https://github.com/user-attachments/assets/c111aa4c-9559-4d74-a2f6-60e322c273d4
9 |
10 | ## Why This is Helpful
11 | While tools like Google Flights work great for simple trips, this tool shines when dealing with complex travel plans. Here's why:
12 |
13 | - **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
14 | - **Flexible Date Search**: Easily search across multiple days to find the best prices without manually checking each date
15 | - **Complex Itineraries**: Perfect for multi-city trips, one-stop flights, or when you need to compare different route options you can just ask!
16 | - **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.
17 |
18 | 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.
19 |
20 | ## Features
21 | - Search for flights between multiple destinations
22 | - Support for one-way, round-trip, and multi-city flight queries
23 | - Detailed flight offer information
24 | - Flexible search parameters (departure times, cabin class, number of passengers)
25 | - Automatic handling of flight connections
26 | - Search for flights within multiple days to find the best flight for your trip (slower)
27 | ## Prerequisites
28 | - Python 3.x
29 | - Duffel API Live Key
30 |
31 | ## Getting Your Duffel API Key
32 | 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.
33 |
34 | 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.
35 |
36 | ### Test Mode First (Recommended)
37 | 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:
38 | 1. Visit [Duffel's registration page](https://app.duffel.com/join)
39 | 2. Create an account (you can select "Personal Use" for Company Name)
40 | 3. Navigate to More > Developer to find your test API key (one is already provided)
41 |
42 | ### Getting a Live API Key
43 | To access real flight data, follow these steps:
44 | 1. In the Duffel dashboard, toggle "Test Mode" off in the top left corner
45 | 2. The verification process requires multiple steps - you'll need to toggle test mode off repeatedly:
46 | - First toggle: Verify your email address
47 | - Toggle again: Complete company information (Personal Use is fine)
48 | - Toggle again: Add payment information (required by Duffel but NO CHARGES will be made by this MCP server)
49 | - Toggle again: Complete any remaining verification steps
50 | - Final toggle: Access live mode after clicking "Agree and Submit"
51 | 3. Once fully verified, go to More > Developer > Create Live Token
52 | 4. Copy your live API key
53 |
54 | 💡 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.
55 |
56 | ⚠️ IMPORTANT NOTES:
57 | - Your payment information is handled directly by Duffel and is not accessed or stored by the MCP server
58 | - This MCP server is READ-ONLY - it can only search for flights, not book them
59 | - No charges will be made to your payment method through this integration
60 | - All sensitive information (including API keys) stays local to your machine
61 | - You can start with the test API key (`duffel_test`) to evaluate the functionality
62 | - The verification process may take some time - this is a standard Duffel requirement
63 |
64 | ### Security Note
65 | 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.
66 |
67 | ### Note on API Usage Limits
68 | - Check Duffel's current pricing and usage limits
69 | - Different tiers available based on your requirements
70 | - Recommended to review current pricing on their website
71 |
72 | ## Installation
73 |
74 | ### Installing via Smithery
75 |
76 | To install Find Flights for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@ravinahp/travel-mcp):
77 |
78 | ```bash
79 | npx -y @smithery/cli install @ravinahp/travel-mcp --client claude
80 | ```
81 |
82 | ### Manual Installation
83 | Clone the repository:
84 | ```bash
85 | git clone https://github.com/ravinahp/flights-mcp
86 | cd flights-mcp
87 | ```
88 |
89 | Install dependencies using uv:
90 | ```bash
91 | uv sync
92 | ```
93 | Note: We use uv instead of pip since the project uses pyproject.toml for dependency management.
94 |
95 | ## Configure as MCP Server
96 | To add this tool as an MCP server, modify your Claude desktop configuration file.
97 |
98 | Configuration file locations:
99 | - MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
100 | - Windows: `%APPDATA%/Claude/claude_desktop_config.json`
101 |
102 | Add the following configuration to your JSON file:
103 | ```json
104 | {
105 | "flights-mcp": {
106 | "command": "uv",
107 | "args": [
108 | "--directory",
109 | "/Users/YOUR_USERNAME/Code/flights-mcp",
110 | "run",
111 | "flights-mcp"
112 | ],
113 | "env": {
114 | "DUFFEL_API_KEY_LIVE": "your_duffel_live_api_key_here"
115 | }
116 | }
117 | }
118 | ```
119 |
120 | ⚠️ IMPORTANT:
121 | - Replace `YOUR_USERNAME` with your actual system username
122 | - Replace `your_duffel_live_api_key_here` with your actual Duffel Live API key
123 | - Ensure the directory path matches your local installation
124 |
125 | ## Deployment
126 | ### Building
127 | Prepare the package:
128 | ```bash
129 | # Sync dependencies and update lockfile
130 | uv sync
131 |
132 | # Build package
133 | uv build
134 | ```
135 | This will create distributions in the `dist/` directory.
136 |
137 | ## Debugging
138 | For the best debugging experience, use the MCP Inspector:
139 | ```bash
140 | npx @modelcontextprotocol/inspector uv --directory /path/to/find-flights-mcp run flights-mcp
141 | ```
142 |
143 | The Inspector provides:
144 | - Real-time request/response monitoring
145 | - Input/output validation
146 | - Error tracking
147 | - Performance metrics
148 |
149 | ## Available Tools
150 |
151 | ### 1. Search Flights
152 | ```python
153 | @mcp.tool()
154 | async def search_flights(params: FlightSearch) -> str:
155 | """Search for flights based on parameters."""
156 | ```
157 | Supports three flight types:
158 | - One-way flights
159 | - Round-trip flights
160 | - Multi-city flights
161 |
162 | Parameters include:
163 | - `type`: Flight type ('one_way', 'round_trip', 'multi_city')
164 | - `origin`: Origin airport code
165 | - `destination`: Destination airport code
166 | - `departure_date`: Departure date (YYYY-MM-DD)
167 | - Optional parameters:
168 | - `return_date`: Return date for round-trips
169 | - `adults`: Number of adult passengers
170 | - `cabin_class`: Preferred cabin class
171 | - `departure_time`: Specific departure time range
172 | - `arrival_time`: Specific arrival time range
173 | - `max_connections`: Maximum number of connections
174 |
175 | ### 2. Get Offer Details
176 | ```python
177 | @mcp.tool()
178 | async def get_offer_details(params: OfferDetails) -> str:
179 | """Get detailed information about a specific flight offer."""
180 | ```
181 | Retrieves comprehensive details for a specific flight offer using its unique ID.
182 |
183 | ### 3. Search Multi-City Flights
184 | ```python
185 | @mcp.tool(name="search_multi_city")
186 | async def search_multi_city(params: MultiCityRequest) -> str:
187 | """Search for multi-city flights."""
188 | ```
189 | Specialized tool for complex multi-city flight itineraries.
190 |
191 | Parameters include:
192 | - `segments`: List of flight segments
193 | - `adults`: Number of adult passengers
194 | - `cabin_class`: Preferred cabin class
195 | - `max_connections`: Maximum number of connections
196 |
197 | ## Use Cases
198 | ### Some Example (But try it out yourself!)
199 | You can use these tools to find flights with various complexities:
200 | - "Find a one-way flight from SFO to NYC on Jan 7 for 2 adults in business class"
201 | - "Search for a round-trip flight from LAX to London, departing Jan 8 and returning Jan 15"
202 | - "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"
203 | - "What is the cheapest flight from SFO to LAX from Jan 7 to Jan 15 for 2 adults in economy class?"
204 | - 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"
205 |
206 | ## Response Format
207 | The tools return JSON-formatted responses with:
208 | - Flight offer details
209 | - Pricing information
210 | - Slice (route) details
211 | - Carrier information
212 | - Connection details
213 |
214 | ## Error Handling
215 | The service includes robust error handling for:
216 | - API request failures
217 | - Invalid airport codes
218 | - Missing or invalid API keys
219 | - Network timeouts
220 | - Invalid search parameters
221 |
222 | ## Contributing
223 | [Add guidelines for contribution, if applicable]
224 |
225 | ## License
226 |
227 | This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
228 |
229 | ## Performance Notes
230 | - Searches are limited to 50 offers for one-way/round-trip flights
231 | - Multi-city searches are limited to 10 offers
232 | - Supplier timeout is set to 15-30 seconds depending on the search type
233 |
234 | ### Cabin Classes
235 | Available cabin classes:
236 | - `economy`: Standard economy class
237 | - `premium_economy`: Premium economy class
238 | - `business`: Business class
239 | - `first`: First class
240 |
241 | Example request with cabin class:
242 | ```json
243 | {
244 | "params": {
245 | "type": "one_way",
246 | "adults": 1,
247 | "origin": "SFO",
248 | "destination": "LAX",
249 | "departure_date": "2025-01-12",
250 | "cabin_class": "business" // Specify desired cabin class
251 | }
252 | }
253 | ```
254 |
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
1 | # Tests package initialization
```
--------------------------------------------------------------------------------
/src/flights/api/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Duffel API client package."""
2 |
3 | from .client import DuffelClient
4 |
5 | __all__ = ['DuffelClient']
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
1 | [pytest]
2 | asyncio_mode = auto
3 | testpaths = tests
4 | python_files = test_*.py
5 | pythonpath = src
6 | log_cli = true
7 | log_cli_level = INFO
```
--------------------------------------------------------------------------------
/src/flights/config/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Configuration package."""
2 |
3 | from .api import DUFFEL_API_URL, DUFFEL_API_VERSION, get_api_token
4 |
5 | __all__ = ['DUFFEL_API_URL', 'DUFFEL_API_VERSION', 'get_api_token']
```
--------------------------------------------------------------------------------
/src/flights/services/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Flight search services."""
2 |
3 | from .search import search_flights, get_offer_details, search_multi_city
4 |
5 | __all__ = ['search_flights', 'get_offer_details', 'search_multi_city']
```
--------------------------------------------------------------------------------
/src/flights/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """Flight search MCP package initialization."""
2 |
3 | from . import server
4 | import asyncio
5 |
6 | def main():
7 | """Main entry point for the package."""
8 | asyncio.run(server.main())
9 |
10 | __all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/src/flights/models/offers.py:
--------------------------------------------------------------------------------
```python
1 | """Offer-related models."""
2 |
3 | from pydantic import BaseModel, Field
4 |
5 | class OfferDetails(BaseModel):
6 | """Model for getting detailed offer information."""
7 | offer_id: str = Field(..., description="The ID of the offer to get details for")
```
--------------------------------------------------------------------------------
/src/flights/models/flight_search.py:
--------------------------------------------------------------------------------
```python
1 | """Flight search models."""
2 |
3 | from .search import FlightSearch
4 | from .multi_city import MultiCityRequest
5 | from .segments import FlightSegment
6 | from .offers import OfferDetails
7 |
8 | __all__ = [
9 | 'FlightSearch',
10 | 'MultiCityRequest',
11 | 'FlightSegment',
12 | 'OfferDetails',
13 | ]
```
--------------------------------------------------------------------------------
/src/flights/models/time_specs.py:
--------------------------------------------------------------------------------
```python
1 | """Time specification models."""
2 |
3 | from pydantic import BaseModel, Field
4 |
5 | class TimeSpec(BaseModel):
6 | """Model for time range specification."""
7 | from_time: str = Field(..., description="Start time (HH:MM)", pattern="^([0-1]?[0-9]|2[0-3]):[0-5][0-9]$")
8 | 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
1 | """Flight segment models."""
2 |
3 | from pydantic import BaseModel, Field
4 |
5 | class FlightSegment(BaseModel):
6 | """Model for a single flight segment in a multi-city trip."""
7 | origin: str = Field(..., description="Origin airport code")
8 | destination: str = Field(..., description="Destination airport code")
9 | departure_date: str = Field(..., description="Departure date (YYYY-MM-DD)")
```
--------------------------------------------------------------------------------
/src/flights/config/api.py:
--------------------------------------------------------------------------------
```python
1 | """Duffel API configuration."""
2 |
3 | import os
4 | from typing import Final
5 |
6 | # API Constants
7 | DUFFEL_API_URL: Final = "https://api.duffel.com"
8 | DUFFEL_API_VERSION: Final = "v2"
9 |
10 | def get_api_token() -> str:
11 | """Get Duffel API token from environment."""
12 | token = os.getenv("DUFFEL_API_KEY_LIVE")
13 | if not token:
14 | raise ValueError("DUFFEL_API_KEY_LIVE environment variable not set")
15 | return token
```
--------------------------------------------------------------------------------
/src/flights/server.py:
--------------------------------------------------------------------------------
```python
1 | """Server initialization for find-flights MCP."""
2 |
3 | import logging
4 | from .services.search import mcp
5 |
6 | # Set up logging
7 | logger = logging.getLogger(__name__)
8 |
9 | def main():
10 | """Entry point for the find-flights-mcp application."""
11 | logger.info("Starting Find Flights MCP server")
12 | try:
13 | mcp.run(transport='stdio')
14 | logger.info("Server initialized successfully")
15 | except Exception as e:
16 | logger.error(f"Server error occurred: {str(e)}", exc_info=True)
17 | raise
18 |
19 | if __name__ == "__main__":
20 | main()
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "flights-mcp"
3 | version = "0.1.0"
4 | description = "Flight search MCP server using Duffel API"
5 | requires-python = ">=3.10"
6 | dependencies = [
7 | "httpx",
8 | "python-dotenv",
9 | "pydantic",
10 | "mcp",
11 | ]
12 | license = "MIT"
13 |
14 | [project.scripts]
15 | flights-mcp = "flights:main"
16 |
17 | [build-system]
18 | requires = ["hatchling"]
19 | build-backend = "hatchling.build"
20 |
21 | [tool.hatch.build.targets.wheel]
22 | packages = ["src/flights"]
23 |
24 | [tool.pytest.ini_options]
25 | asyncio_mode = "auto"
26 | testpaths = ["tests"]
27 | python_files = ["test_*.py"]
28 | pythonpath = ["src"]
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
1 | # Smithery configuration file: https://smithery.ai/docs/deployments
2 |
3 | startCommand:
4 | type: stdio
5 | configSchema:
6 | # JSON Schema defining the configuration options for the MCP.
7 | type: object
8 | required:
9 | - duffelApiKeyLive
10 | properties:
11 | duffelApiKeyLive:
12 | type: string
13 | description: The live API key for accessing the Duffel flight search service.
14 | commandFunction:
15 | # A function that produces the CLI command to start the MCP on stdio.
16 | |-
17 | (config) => ({ command: 'flights-mcp', env: { DUFFEL_API_KEY_LIVE: config.duffelApiKeyLive } })
```
--------------------------------------------------------------------------------
/src/flights/models/multi_city.py:
--------------------------------------------------------------------------------
```python
1 | """Multi-city flight search models."""
2 |
3 | from typing import Optional, List, Literal
4 | from pydantic import BaseModel, Field
5 | from .time_specs import TimeSpec
6 | from .segments import FlightSegment
7 |
8 | class MultiCityRequest(BaseModel):
9 | """Model for multi-city flight search."""
10 | type: Literal["multi_city"]
11 | segments: List[FlightSegment] = Field(..., min_items=2, description="Flight segments")
12 | cabin_class: str = Field("economy", description="Cabin class")
13 | adults: int = Field(1, description="Number of adult passengers")
14 | max_connections: int = Field(None, description="Maximum number of connections (0 for non-stop)")
15 | departure_time: TimeSpec | None = Field(None, description="Optional departure time range")
16 | arrival_time: TimeSpec | None = Field(None, description="Optional arrival time range")
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | # Use a Python image with uv pre-installed
2 | FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
3 |
4 | # Install the project into /app
5 | WORKDIR /app
6 |
7 | # Enable bytecode compilation
8 | ENV UV_COMPILE_BYTECODE=1
9 |
10 | # Copy from the cache instead of linking since it's a mounted volume
11 | ENV UV_LINK_MODE=copy
12 |
13 | # Install the project's dependencies using the lockfile and settings
14 | COPY pyproject.toml uv.lock /app/
15 | RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-install-project --no-dev --no-editable
16 |
17 | # Then, add the rest of the project source code and install it
18 | # Installing separately from its dependencies allows optimal layer caching
19 | ADD src /app/src
20 | RUN --mount=type=cache,target=/root/.cache/uv uv sync --frozen --no-dev --no-editable
21 |
22 | FROM python:3.12-slim-bookworm
23 |
24 | WORKDIR /app
25 |
26 | COPY --from=uv /root/.local /root/.local
27 | COPY --from=uv --chown=app:app /app/.venv /app/.venv
28 |
29 | # Place executables in the environment at the front of the path
30 | ENV PATH="/app/.venv/bin:$PATH"
31 |
32 | # Define environment variable for Duffel API key
33 | ENV DUFFEL_API_KEY_LIVE=your_duffel_live_api_key_here
34 |
35 | # Start the MCP server
36 | ENTRYPOINT ["flights-mcp"]
37 |
```
--------------------------------------------------------------------------------
/src/flights/models/search.py:
--------------------------------------------------------------------------------
```python
1 | """Flight search models."""
2 |
3 | from typing import Optional, List
4 | from pydantic import BaseModel, Field
5 | from .time_specs import TimeSpec
6 |
7 | class FlightSearch(BaseModel):
8 | """Model for flight search parameters."""
9 | type: str = Field(..., description="Type of flight: 'one_way', 'round_trip', or 'multi_city'")
10 | origin: str = Field(..., description="Origin airport code")
11 | destination: str = Field(..., description="Destination airport code")
12 | departure_date: str = Field(..., description="Departure date (YYYY-MM-DD)")
13 | return_date: str | None = Field(None, description="Return date for round trips (YYYY-MM-DD)")
14 | departure_time: TimeSpec | None = Field(None, description="Preferred departure time range")
15 | arrival_time: TimeSpec | None = Field(None, description="Preferred arrival time range")
16 | cabin_class: str = Field("economy", description="Cabin class (economy, business, first)")
17 | adults: int = Field(1, description="Number of adult passengers")
18 | max_connections: int = Field(None, description="Maximum number of connections (0 for non-stop)")
19 | additional_stops: Optional[List[dict]] = Field(None, description="Additional stops for multi-city trips")
```
--------------------------------------------------------------------------------
/src/flights/api/client.py:
--------------------------------------------------------------------------------
```python
1 | """Duffel API client."""
2 |
3 | import logging
4 | import httpx
5 | from typing import Dict, Any, List
6 | from ..config import get_api_token
7 | from .endpoints import OfferEndpoints
8 |
9 | class DuffelClient:
10 | """Client for interacting with the Duffel API."""
11 |
12 | def __init__(self, logger: logging.Logger, timeout: float = 30.0):
13 | """Initialize the Duffel API client."""
14 | self.logger = logger
15 | self.timeout = timeout
16 | self._token = get_api_token()
17 | self.base_url = "https://api.duffel.com/air"
18 |
19 | # Headers setup
20 | self.headers = {
21 | "Accept": "application/json",
22 | "Accept-Encoding": "gzip",
23 | "Duffel-Version": "v2",
24 | "Authorization": f"Bearer {self._token}",
25 | "Content-Type": "application/json"
26 | }
27 |
28 | self.logger.info(f"API key starts with: {self._token[:8] if self._token else None}")
29 | self.logger.info(f"Using base URL: {self.base_url}")
30 |
31 | # Initialize endpoints
32 | self.offers = OfferEndpoints(self.base_url, self.headers, self.logger)
33 |
34 | async def __aenter__(self):
35 | """Async context manager entry."""
36 | return self
37 |
38 | async def __aexit__(self, exc_type, exc_val, exc_tb):
39 | """Async context manager exit."""
40 | pass
41 |
42 | async def create_offer_request(self, **kwargs) -> Dict[str, Any]:
43 | """Create an offer request."""
44 | return await self.offers.create_offer_request(**kwargs)
45 |
46 | async def get_offer(self, offer_id: str) -> Dict[str, Any]:
47 | """Get offer details."""
48 | return await self.offers.get_offer(offer_id)
49 |
```
--------------------------------------------------------------------------------
/src/flights/api/endpoints.py:
--------------------------------------------------------------------------------
```python
1 | """Duffel API endpoint handlers."""
2 |
3 | from typing import Dict, Any, List
4 | import logging
5 | import httpx
6 |
7 | class OfferEndpoints:
8 | """Offer-related API endpoints."""
9 |
10 | def __init__(self, base_url: str, headers: Dict, logger: logging.Logger):
11 | self.base_url = base_url
12 | self.headers = headers
13 | self.logger = logger
14 |
15 | async def create_offer_request(
16 | self,
17 | slices: List[Dict],
18 | cabin_class: str = "economy",
19 | adult_count: int = 1,
20 | max_connections: int = None,
21 | return_offers: bool = True,
22 | supplier_timeout: int = 15000
23 | ) -> Dict:
24 | """Create a flight offer request."""
25 | try:
26 | # Format request data
27 | request_data = {
28 | "data": {
29 | "slices": slices,
30 | "passengers": [{"type": "adult"} for _ in range(adult_count)],
31 | "cabin_class": cabin_class,
32 | }
33 | }
34 |
35 | if max_connections is not None:
36 | request_data["data"]["max_connections"] = max_connections
37 |
38 | params = {
39 | "return_offers": str(return_offers).lower(),
40 | "supplier_timeout": supplier_timeout
41 | }
42 |
43 | async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
44 | self.logger.info(f"Creating offer request with data: {request_data}")
45 | response = await client.post(
46 | f"{self.base_url}/offer_requests",
47 | headers=self.headers,
48 | params=params,
49 | json=request_data
50 | )
51 | response.raise_for_status()
52 | data = response.json()
53 |
54 | request_id = data["data"]["id"]
55 | offers = data["data"].get("offers", [])
56 |
57 | self.logger.info(f"Created offer request with ID: {request_id}")
58 | self.logger.info(f"Received {len(offers)} offers")
59 |
60 | return {
61 | "request_id": request_id,
62 | "offers": offers
63 | }
64 |
65 | except Exception as e:
66 | error_msg = f"Error creating offer request: {str(e)}"
67 | self.logger.error(error_msg)
68 | raise
69 |
70 | async def get_offer(self, offer_id: str) -> Dict:
71 | """Get details of a specific offer."""
72 | try:
73 | if not offer_id.startswith("off_"):
74 | raise ValueError("Invalid offer ID format - must start with 'off_'")
75 |
76 | async with httpx.AsyncClient() as client:
77 | response = await client.get(
78 | f"{self.base_url}/offers/{offer_id}",
79 | headers=self.headers
80 | )
81 | response.raise_for_status()
82 | return response.json()
83 | except Exception as e:
84 | self.logger.error(f"Error getting offer {offer_id}: {str(e)}")
85 | raise
```
--------------------------------------------------------------------------------
/tests/test_duffel_api.py:
--------------------------------------------------------------------------------
```python
1 | """Tests for Duffel API client."""
2 |
3 | import pytest
4 | import logging
5 | from datetime import datetime, timedelta
6 | from flights.api import DuffelClient
7 | from flights.models.search import FlightSearch
8 | from flights.models.multi_city import MultiCityRequest
9 |
10 | # Setup logging for tests
11 | logger = logging.getLogger(__name__)
12 |
13 | @pytest.fixture
14 | async def client():
15 | """Create a test client."""
16 | client = DuffelClient(logger)
17 | async with client as c:
18 | yield c
19 |
20 | @pytest.mark.asyncio
21 | async def test_search_one_way(client):
22 | """Test one-way flight search."""
23 | # Get tomorrow's date for testing
24 | tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
25 |
26 | response = await client.create_offer_request(
27 | slices=[{
28 | "origin": "SFO",
29 | "destination": "LAX",
30 | "departure_date": tomorrow
31 | }],
32 | cabin_class="economy",
33 | adult_count=1
34 | )
35 |
36 | assert response is not None
37 | assert "request_id" in response
38 | assert "offers" in response
39 | assert len(response["offers"]) > 0
40 |
41 | @pytest.mark.asyncio
42 | async def test_search_round_trip(client):
43 | """Test round-trip flight search."""
44 | # Get dates for testing
45 | departure = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")
46 | return_date = (datetime.now() + timedelta(days=14)).strftime("%Y-%m-%d")
47 |
48 | response = await client.create_offer_request(
49 | slices=[
50 | {
51 | "origin": "SFO",
52 | "destination": "LAX",
53 | "departure_date": departure
54 | },
55 | {
56 | "origin": "LAX",
57 | "destination": "SFO",
58 | "departure_date": return_date
59 | }
60 | ],
61 | cabin_class="economy",
62 | adult_count=1
63 | )
64 |
65 | assert response is not None
66 | assert "request_id" in response
67 | assert "offers" in response
68 | assert len(response["offers"]) > 0
69 |
70 | @pytest.mark.asyncio
71 | async def test_search_multi_city(client):
72 | """Test multi-city flight search."""
73 | # Get dates for testing
74 | first_date = (datetime.now() + timedelta(days=10)).strftime("%Y-%m-%d")
75 | second_date = (datetime.now() + timedelta(days=15)).strftime("%Y-%m-%d")
76 |
77 | response = await client.create_offer_request(
78 | slices=[
79 | {
80 | "origin": "SFO",
81 | "destination": "LAX",
82 | "departure_date": first_date
83 | },
84 | {
85 | "origin": "LAX",
86 | "destination": "JFK",
87 | "departure_date": second_date
88 | }
89 | ],
90 | cabin_class="economy",
91 | adult_count=1
92 | )
93 |
94 | assert response is not None
95 | assert "request_id" in response
96 | assert "offers" in response
97 |
98 | @pytest.mark.asyncio
99 | async def test_cabin_classes(client):
100 | """Test different cabin classes."""
101 | tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
102 |
103 | cabin_classes = ["economy", "premium_economy", "business", "first"]
104 |
105 | for cabin_class in cabin_classes:
106 | response = await client.create_offer_request(
107 | slices=[{
108 | "origin": "SFO",
109 | "destination": "LAX",
110 | "departure_date": tomorrow
111 | }],
112 | cabin_class=cabin_class,
113 | adult_count=1
114 | )
115 |
116 | assert response is not None
117 | assert "request_id" in response
118 | assert "offers" in response
119 |
120 | @pytest.mark.asyncio
121 | async def test_get_offer(client):
122 | """Test getting offer details."""
123 | # First create an offer request
124 | tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")
125 |
126 | offers_response = await client.create_offer_request(
127 | slices=[{
128 | "origin": "SFO",
129 | "destination": "LAX",
130 | "departure_date": tomorrow
131 | }],
132 | cabin_class="economy",
133 | adult_count=1
134 | )
135 |
136 | assert offers_response is not None
137 | assert "offers" in offers_response
138 | assert len(offers_response["offers"]) > 0
139 |
140 | # Get the first offer's details
141 | offer_id = offers_response["offers"][0]["id"]
142 | offer_details = await client.get_offer(offer_id)
143 |
144 | assert offer_details is not None
145 | assert "data" in offer_details
146 |
147 | @pytest.mark.asyncio
148 | async def test_error_handling(client):
149 | """Test error handling for invalid requests."""
150 | with pytest.raises(Exception):
151 | await client.create_offer_request(
152 | slices=[{
153 | "origin": "INVALID",
154 | "destination": "ALSO_INVALID",
155 | "departure_date": "2025-01-01"
156 | }],
157 | cabin_class="economy",
158 | adult_count=1
159 | )
160 |
161 | @pytest.mark.asyncio
162 | async def test_invalid_offer_id(client):
163 | """Test error handling for invalid offer ID."""
164 | with pytest.raises(ValueError):
165 | await client.get_offer("invalid_offer_id")
```
--------------------------------------------------------------------------------
/src/flights/services/search.py:
--------------------------------------------------------------------------------
```python
1 | """Flight search tools using Duffel API."""
2 |
3 | import logging
4 | from typing import Dict
5 | import json
6 | from mcp.server.fastmcp import FastMCP
7 |
8 | # Import all models through flight_search
9 | from ..models.flight_search import (
10 | FlightSearch,
11 | MultiCityRequest,
12 | OfferDetails
13 | )
14 | from ..models.time_specs import TimeSpec
15 | from ..api import DuffelClient
16 |
17 | # Set up logging
18 | logger = logging.getLogger(__name__)
19 |
20 | # Initialize FastMCP server and API client
21 | mcp = FastMCP("find-flights-mcp")
22 | flight_client = DuffelClient(logger)
23 |
24 |
25 | def _create_slice(origin: str, destination: str, date: str,
26 | departure_time: TimeSpec | None = None,
27 | arrival_time: TimeSpec | None = None) -> Dict:
28 | """Helper to create a slice with time ranges."""
29 | slice_data = {
30 | "origin": origin,
31 | "destination": destination,
32 | "departure_date": date,
33 | "departure_time": {
34 | "from": "00:00",
35 | "to": "23:59"
36 | },
37 | "arrival_time": {
38 | "from": "00:00",
39 | "to": "23:59"
40 | }
41 | }
42 |
43 | if departure_time:
44 | slice_data["departure_time"] = {
45 | "from": departure_time.from_time,
46 | "to": departure_time.to_time
47 | }
48 |
49 | if arrival_time:
50 | slice_data["arrival_time"] = {
51 | "from": arrival_time.from_time,
52 | "to": arrival_time.to_time
53 | }
54 |
55 | return slice_data
56 |
57 | @mcp.tool()
58 | async def search_flights(params: FlightSearch) -> str:
59 | """Search for flights based on parameters."""
60 | try:
61 | slices = []
62 |
63 | # Build slices based on flight type
64 | if params.type == "one_way":
65 | slices = [_create_slice(
66 | params.origin,
67 | params.destination,
68 | params.departure_date,
69 | params.departure_time,
70 | params.arrival_time
71 | )]
72 | elif params.type == "round_trip":
73 | if not params.return_date:
74 | raise ValueError("Return date required for round-trip flights")
75 | slices = [
76 | _create_slice(
77 | params.origin,
78 | params.destination,
79 | params.departure_date,
80 | params.departure_time,
81 | params.arrival_time
82 | ),
83 | _create_slice(
84 | params.destination,
85 | params.origin,
86 | params.return_date,
87 | params.departure_time,
88 | params.arrival_time
89 | )
90 | ]
91 | elif params.type == "multi_city":
92 | if not params.additional_stops:
93 | raise ValueError("Additional stops required for multi-city flights")
94 |
95 | # First leg
96 | slices.append({
97 | "origin": params.origin,
98 | "destination": params.destination,
99 | "departure_date": params.departure_date,
100 | "departure_time": {
101 | "from": "00:00",
102 | "to": "23:59"
103 | },
104 | "arrival_time": {
105 | "from": "00:00",
106 | "to": "23:59"
107 | }
108 | })
109 |
110 | # Additional legs
111 | for stop in params.additional_stops:
112 | slices.append({
113 | "origin": stop["origin"],
114 | "destination": stop["destination"],
115 | "departure_date": stop["departure_date"],
116 | "departure_time": {
117 | "from": "00:00",
118 | "to": "23:59"
119 | },
120 | "arrival_time": {
121 | "from": "00:00",
122 | "to": "23:59"
123 | }
124 | })
125 |
126 | # Use async context manager
127 | async with flight_client as client:
128 | response = await client.create_offer_request(
129 | slices=slices,
130 | cabin_class=params.cabin_class,
131 | adult_count=params.adults,
132 | max_connections=params.max_connections,
133 | return_offers=True,
134 | supplier_timeout=15000
135 | )
136 |
137 | # Format the response
138 | formatted_response = {
139 | 'request_id': response['request_id'],
140 | 'offers': []
141 | }
142 |
143 | # Get all offers (limit to 10 to manage response size)
144 | for offer in response.get('offers', [])[:50]: # Keep the slice to limit offers
145 | offer_details = {
146 | 'offer_id': offer.get('id'),
147 | 'price': {
148 | 'amount': offer.get('total_amount'),
149 | 'currency': offer.get('total_currency')
150 | },
151 | 'slices': []
152 | }
153 |
154 | # Only include essential slice details
155 | for slice in offer.get('slices', []):
156 | segments = slice.get('segments', [])
157 | if segments: # Check if there are any segments
158 | slice_details = {
159 | 'origin': slice['origin']['iata_code'],
160 | 'destination': slice['destination']['iata_code'],
161 | 'departure': segments[0].get('departing_at'), # First segment departure
162 | 'arrival': segments[-1].get('arriving_at'), # Last segment arrival
163 | 'duration': slice.get('duration'),
164 | 'carrier': segments[0].get('marketing_carrier', {}).get('name'),
165 | 'stops': len(segments) - 1,
166 | 'stops_description': 'Non-stop' if len(segments) == 1 else f'{len(segments) - 1} stop{"s" if len(segments) - 1 > 1 else ""}',
167 | 'connections': []
168 | }
169 |
170 | # Add connection information if there are multiple segments
171 | if len(segments) > 1:
172 | for i in range(len(segments)-1):
173 | connection = {
174 | 'airport': segments[i].get('destination', {}).get('iata_code'),
175 | 'arrival': segments[i].get('arriving_at'),
176 | 'departure': segments[i+1].get('departing_at'),
177 | 'duration': segments[i+1].get('duration')
178 | }
179 | slice_details['connections'].append(connection)
180 |
181 | offer_details['slices'].append(slice_details)
182 |
183 | formatted_response['offers'].append(offer_details)
184 |
185 | return json.dumps(formatted_response, indent=2)
186 |
187 | except Exception as e:
188 | logger.error(f"Error searching flights: {str(e)}", exc_info=True)
189 | raise
190 |
191 | @mcp.tool()
192 | async def get_offer_details(params: OfferDetails) -> str:
193 | """Get detailed information about a specific flight offer."""
194 | try:
195 | async with flight_client as client:
196 | response = await client.get_offer(
197 | offer_id=params.offer_id
198 | )
199 | return json.dumps(response, indent=2)
200 |
201 | except Exception as e:
202 | logger.error(f"Error getting offer details: {str(e)}", exc_info=True)
203 | raise
204 |
205 | @mcp.tool(name="search_multi_city")
206 | async def search_multi_city(params: MultiCityRequest) -> str:
207 | """Search for multi-city flights."""
208 | try:
209 | slices = []
210 | for segment in params.segments:
211 | slices.append(_create_slice(
212 | segment.origin,
213 | segment.destination,
214 | segment.departure_date,
215 | None,
216 | None
217 | ))
218 |
219 | # Use async context manager with shorter timeout
220 | async with flight_client as client:
221 | response = await client.create_offer_request(
222 | slices=slices,
223 | cabin_class=params.cabin_class,
224 | adult_count=params.adults,
225 | max_connections=params.max_connections,
226 | return_offers=True,
227 | supplier_timeout=30000 # Increased timeout for multi-city
228 | )
229 |
230 | # Format response inside the context
231 | formatted_response = {
232 | 'request_id': response['request_id'],
233 | 'offers': []
234 | }
235 |
236 | # Process offers inside the context
237 | for offer in response.get('offers', [])[:10]:
238 | offer_details = {
239 | 'offer_id': offer.get('id'),
240 | 'price': {
241 | 'amount': offer.get('total_amount'),
242 | 'currency': offer.get('total_currency')
243 | },
244 | 'slices': []
245 | }
246 |
247 | for slice in offer.get('slices', []):
248 | segments = slice.get('segments', [])
249 | if segments:
250 | slice_details = {
251 | 'origin': slice['origin']['iata_code'],
252 | 'destination': slice['destination']['iata_code'],
253 | 'departure': segments[0].get('departing_at'),
254 | 'arrival': segments[-1].get('arriving_at'),
255 | 'duration': slice.get('duration'),
256 | 'carrier': segments[0].get('marketing_carrier', {}).get('name'),
257 | 'stops': len(segments) - 1,
258 | 'stops_description': 'Non-stop' if len(segments) == 1 else f'{len(segments) - 1} stop{"s" if len(segments) - 1 > 1 else ""}',
259 | 'connections': []
260 | }
261 |
262 | if len(segments) > 1:
263 | for i in range(len(segments)-1):
264 | connection = {
265 | 'airport': segments[i].get('destination', {}).get('iata_code'),
266 | 'arrival': segments[i].get('arriving_at'),
267 | 'departure': segments[i+1].get('departing_at'),
268 | 'duration': segments[i+1].get('duration')
269 | }
270 | slice_details['connections'].append(connection)
271 |
272 | offer_details['slices'].append(slice_details)
273 |
274 | formatted_response['offers'].append(offer_details)
275 |
276 | return json.dumps(formatted_response, indent=2)
277 |
278 | except Exception as e:
279 | logger.error(f"Error searching flights: {str(e)}", exc_info=True)
280 | raise
```