from fastmcp import FastMCP
from typing import Annotated
import requests
from dotenv import load_dotenv
import os
load_dotenv()
app = FastMCP("maersk-mcp-server")
headers = {"Consumer-Key": os.getenv("CONSUMER_KEY")}
@app.tool()
async def get_vessel_imo(
vessel_name: Annotated[str, "The name of the vessel"]
) -> str:
"""
Get the IMO number of a vessel from the Maersk vessel API.
"""
base_url = "https://api.maersk.com/reference-data/vessels?vesselNames="
resp = requests.get(base_url + vessel_name, headers=headers)
try:
resp.raise_for_status()
data = resp.json()
return data[0]["vesselIMONumber"]
except requests.exceptions.RequestException as e:
return f"Error: {e}"
@app.tool()
async def get_vessel_schedule(
vessel_imo: Annotated[str, "The IMO number of the vessel"],
voyage_number: Annotated[str, "The voyage number of the vessel"],
port_of_loading: Annotated[str, "The port of loading"],
iso_country_code: Annotated[str, "The ISO country code of the port of loading"],
) -> str:
"""
Get the schedule of a vessel from the Maersk vessel API, by passing the IMO number, voyage number, port of loading and ISO country code.
If the vessel IMO is not provided and the name instead, fetch the IMO number using the get_vessel_imo tool and then proceed.
"""
base_url = f"https://api.maersk.com/shipment-deadlines?ISOCountryCode={iso_country_code}&portOfLoad={port_of_loading}&vesselIMONumber={vessel_imo}&voyage={voyage_number}"
resp = requests.get(base_url, headers=headers)
try:
resp.raise_for_status()
data = resp.json()
terminal_name = data[0]["shipmentDeadlines"]["terminalName"]
deadlines_joined = ", ".join(f'{deadline["deadlineName"]} on {deadline["deadlineLocal"]}' for deadline in data[0]["shipmentDeadlines"]["deadlines"])
return f"The vessel {vessel_imo} is scheduled to arrive at {terminal_name} with deadlines: {deadlines_joined}."
except requests.exceptions.RequestException as e:
return f"Error: {e}"
@app.tool()
async def get_port_calls(
country_code: Annotated[str, "The ISO country code of the port of loading"],
city_name: Annotated[str, "The name of the loading port city"]
) -> str:
"""
Get the port calls of a vessel from the Maersk vessel API, by passing the ISO country code, city name and the date range.
This can be used to obtain which vessels are scheduled to call at a port in a given date range.
"""
date_range = "P1W"
carrier_code = "MAEU"
base_url = f"https://api.maersk.com/schedules/port-calls?carrierCodes={carrier_code}&countryCode={country_code}&cityName={city_name}&dateRange={date_range}"
resp = requests.get(base_url, headers=headers)
try:
resp.raise_for_status()
data = resp.json()
output_lines = []
for facility_entry in data["portCalls"]:
facility_name = facility_entry["facility"]["locationName"]
output_lines.append(f"Facility: {facility_name}")
for call in facility_entry["facilityCalls"]:
vessel = call["transport"]["vessel"]
voyage = call["transport"]["outboundService"]
vessel_name = vessel["vesselName"]
vessel_imo = vessel["vesselIMONumber"]
voyage_number = voyage["carrierVoyageNumber"]
output_lines.append(f" - {vessel_name} (IMO: {vessel_imo}, Voyage: {voyage_number})")
result_string = "\n".join(output_lines)
return result_string
except requests.exceptions.RequestException as e:
return f"Error: {e}"
