API Reference

surf_spot_finder.cli

find_surf_spot(config_file=None) async

Find the best surf spot based on the given criteria.

Parameters:

Name	Type	Description	Default
config_file	str | None	Path to a YAML config file. See Config	None

Source code in src/surf_spot_finder/cli.py

async def find_surf_spot(
    config_file: str | None = None,
) -> str:
    """Find the best surf spot based on the given criteria.

    Args:
        config_file: Path to a YAML config file.
            See [Config][surf_spot_finder.config.Config]

    """
    if config_file is None:
        config = Config.from_dict({})
    else:
        logger.info("Loading %s", config_file)
        config = Config.from_yaml(config_file)

    if not config.main_agent.instructions:
        if config.framework == AgentFramework.SMOLAGENTS:
            config.main_agent.instructions = SYSTEM_PROMPT
        elif config.framework == AgentFramework.OPENAI:
            config.main_agent.instructions = SINGLE_AGENT_SYSTEM_PROMPT

    logger.info("Loading %s agent", config.framework)
    logger.info("Managed agents: %s", config.managed_agents)
    agent = await AnyAgent.create_async(
        agent_framework=config.framework,
        agent_config=config.main_agent,
        managed_agents=config.managed_agents,
        tracing=TracingConfig(console=True, cost_info=True),
    )

    query = config.input_prompt_template.format(
        LOCATION=config.location,
        MAX_DRIVING_HOURS=config.max_driving_hours,
        DATE=config.date,
    )
    logger.info("Running agent with query:\n%s", query)
    agent_trace = await agent.run_async(query)

    logger.info("Final output from agent:\n%s", agent_trace.final_output)

    # dump the trace in the "output" directory
    output_dir = "output"
    os.makedirs(output_dir, exist_ok=True)
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    file_path = Path(output_dir) / f"{timestamp}_trace.json"
    with open(file_path, "w") as f:
        f.write(agent_trace.model_dump_json(indent=2))

    if config.evaluation_cases is not None:
        results = []
        logger.info("Found evaluation cases, running trace evaluation")
        for i, case in enumerate(config.evaluation_cases):
            logger.info("Evaluating case: %s", case)
            result: TraceEvaluationResult = evaluate(
                evaluation_case=case,
                trace=agent_trace,
                agent_framework=config.framework,
            )
            for list_of_checkpoints in [
                result.checkpoint_results,
                result.direct_results,
                result.hypothesis_answer_results,
            ]:
                for checkpoint in list_of_checkpoints:
                    msg = (
                        f"Checkpoint: {checkpoint.criteria}\n"
                        f"\tPassed: {checkpoint.passed}\n"
                        f"\tReason: {checkpoint.reason}\n"
                        f"\tScore: {'%d/%d' % (checkpoint.points, checkpoint.points) if checkpoint.passed else '0/%d' % checkpoint.points}"
                    )
                    logger.info(msg)
            logger.info("==========================")
            logger.info("Overall Score: %d%%", 100 * result.score)
            logger.info("==========================")
            results.append(result)
        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
        file_path = Path(output_dir) / f"{timestamp}_eval_case_{i}.json"
        with open(file_path, "w") as f:
            f.write(result.model_dump_json(indent=2))

surf_spot_finder.config.Config

Bases: BaseModel

Source code in src/surf_spot_finder/config.py

class Config(BaseModel):
    model_config = ConfigDict(extra="forbid")

    location: str
    max_driving_hours: PositiveInt
    date: FutureDatetime
    input_prompt_template: Annotated[str, AfterValidator(validate_prompt)] = (
        INPUT_PROMPT_TEMPLATE
    )

    framework: AgentFramework

    main_agent: AgentConfig
    managed_agents: list[AgentConfig] | None = None

    evaluation_cases: list[EvaluationCase] | None = None

    @classmethod
    def from_dict(cls, data: dict) -> "Config":
        """
        Create a Config instance from a dictionary.
        Args:
            data (dict): A dictionary containing the configuration data.

        Returns:
            Config: A new Config instance populated with values from the dictionary.
        """
        # for each tool listed in main_agent.tools, use import lib to import it and replace the str with the callable
        callables = []
        if data.get("main_agent") is None:
            data["main_agent"] = {}
        if not data["main_agent"].get("model_id"):
            data["main_agent"]["model_id"] = get_litellm_model_id("main_agent")
        else:
            logger.info(f"Main agent using model_id {data['main_agent']['model_id']}")
        for tool in data["main_agent"].get("tools", []):
            if isinstance(tool, str):
                module_name, func_name = tool.rsplit(".", 1)
                module = __import__(module_name, fromlist=[func_name])
                callables.append(getattr(module, func_name))
            else:
                # this means it must be an MCPStdioParams
                # For the purposes of this demo, currently we just look for the filesystem MCP which we have a placeholder
                # for the path variable (which controls which dirs the MCP will have access to).
                mcp_tool = set_mcp_settings(tool)
                callables.append(mcp_tool)
        data["main_agent"]["tools"] = callables
        for agent in data.get("managed_agents", []):
            if agent.get("model_id") is None:
                agent["model_id"] = get_litellm_model_id(
                    agent.get("name", "managed_agent")
                )
            else:
                logger.info(f"Agent {agent['name']} using model_id {agent['model_id']}")
            callables = []
            for tool in agent.get("tools", []):
                if isinstance(tool, str):
                    module_name, func_name = tool.rsplit(".", 1)
                    module = __import__(module_name, fromlist=[func_name])
                    callables.append(getattr(module, func_name))
                else:
                    # this means it must be an MCPStdioParams
                    mcp_tool = set_mcp_settings(tool)
                    callables.append(mcp_tool)
            agent["tools"] = callables
        if not data.get("framework"):
            data["framework"] = ask_framework()
        else:
            logger.info(f"Using framework {data['framework']}")
        if not data.get("location"):
            data["location"] = location_picker()
        else:
            logger.info(f"Using location {data['location']}")
        if not data.get("max_driving_hours"):
            data["max_driving_hours"] = max_driving_hours_picker()
        else:
            logger.info(f"Using max driving hours {data['max_driving_hours']}")
        if not data.get("date"):
            data["date"] = date_picker()
        else:
            logger.info(f"Using date {data['date']}")

        return cls(**data)

    @classmethod
    def from_yaml(cls, yaml_path: str) -> "Config":
        """
        with open(yaml_path, "r") as f:
            data = yaml.safe_load(f)
        return cls(**data)    yaml_path: Path to the YAML configuration file

        Returns:
            Config: A new Config instance populated with values from the YAML file
        """
        with open(yaml_path, "r") as f:
            data = yaml.safe_load(f)
        return cls.from_dict(data)

from_dict(data) classmethod

Create a Config instance from a dictionary. Args: data (dict): A dictionary containing the configuration data.

Returns:

Name	Type	Description
Config	Config	A new Config instance populated with values from the dictionary.

Source code in src/surf_spot_finder/config.py

@classmethod
def from_dict(cls, data: dict) -> "Config":
    """
    Create a Config instance from a dictionary.
    Args:
        data (dict): A dictionary containing the configuration data.

    Returns:
        Config: A new Config instance populated with values from the dictionary.
    """
    # for each tool listed in main_agent.tools, use import lib to import it and replace the str with the callable
    callables = []
    if data.get("main_agent") is None:
        data["main_agent"] = {}
    if not data["main_agent"].get("model_id"):
        data["main_agent"]["model_id"] = get_litellm_model_id("main_agent")
    else:
        logger.info(f"Main agent using model_id {data['main_agent']['model_id']}")
    for tool in data["main_agent"].get("tools", []):
        if isinstance(tool, str):
            module_name, func_name = tool.rsplit(".", 1)
            module = __import__(module_name, fromlist=[func_name])
            callables.append(getattr(module, func_name))
        else:
            # this means it must be an MCPStdioParams
            # For the purposes of this demo, currently we just look for the filesystem MCP which we have a placeholder
            # for the path variable (which controls which dirs the MCP will have access to).
            mcp_tool = set_mcp_settings(tool)
            callables.append(mcp_tool)
    data["main_agent"]["tools"] = callables
    for agent in data.get("managed_agents", []):
        if agent.get("model_id") is None:
            agent["model_id"] = get_litellm_model_id(
                agent.get("name", "managed_agent")
            )
        else:
            logger.info(f"Agent {agent['name']} using model_id {agent['model_id']}")
        callables = []
        for tool in agent.get("tools", []):
            if isinstance(tool, str):
                module_name, func_name = tool.rsplit(".", 1)
                module = __import__(module_name, fromlist=[func_name])
                callables.append(getattr(module, func_name))
            else:
                # this means it must be an MCPStdioParams
                mcp_tool = set_mcp_settings(tool)
                callables.append(mcp_tool)
        agent["tools"] = callables
    if not data.get("framework"):
        data["framework"] = ask_framework()
    else:
        logger.info(f"Using framework {data['framework']}")
    if not data.get("location"):
        data["location"] = location_picker()
    else:
        logger.info(f"Using location {data['location']}")
    if not data.get("max_driving_hours"):
        data["max_driving_hours"] = max_driving_hours_picker()
    else:
        logger.info(f"Using max driving hours {data['max_driving_hours']}")
    if not data.get("date"):
        data["date"] = date_picker()
    else:
        logger.info(f"Using date {data['date']}")

    return cls(**data)

from_yaml(yaml_path) classmethod

with open(yaml_path, "r") as f: data = yaml.safe_load(f) return cls(**data) yaml_path: Path to the YAML configuration file

Returns:

Name	Type	Description
Config	Config	A new Config instance populated with values from the YAML file

Source code in src/surf_spot_finder/config.py

@classmethod
def from_yaml(cls, yaml_path: str) -> "Config":
    """
    with open(yaml_path, "r") as f:
        data = yaml.safe_load(f)
    return cls(**data)    yaml_path: Path to the YAML configuration file

    Returns:
        Config: A new Config instance populated with values from the YAML file
    """
    with open(yaml_path, "r") as f:
        data = yaml.safe_load(f)
    return cls.from_dict(data)

surf_spot_finder.no_framework

find_surf_spot_no_framework(location, max_driving_hours, date, model_id)

Find the best surf spot based on the given location and date.

Uses the following tools:

• any_agent.tools.web_browsing
• surf_spot_finder.tools.openmeteo
• surf_spot_finder.tools.openstreetmap

To find nearby spots along with the forecast and recommended conditions for the spot.

Then, uses litellm with the provided model_id to score each spot based on the available information.

Parameters:

Name	Type	Description	Default
location	str	The place of interest.	required
max_driving_hours	int	Used to limit the surf spots based on the distance to location.	required
date	datetime	Used to filter the forecast results.	required
model_id	str	Can be any of the litellm providers.	required

Returns:

Type	Description
list[SpotScore]	A list of spot scores and reasons for the value.

Source code in src/surf_spot_finder/no_framework.py

def find_surf_spot_no_framework(
    location: str, max_driving_hours: int, date: datetime, model_id: str
) -> list[SpotScore]:
    """Find the best surf spot based on the given `location` and `date`.

    Uses the following tools:

    - any_agent.tools.web_browsing
    - [surf_spot_finder.tools.openmeteo][]
    - [surf_spot_finder.tools.openstreetmap][]

    To find nearby spots along with the forecast and
    recommended conditions for the spot.

    Then, uses `litellm` with the provided `model_id` to score
    each spot based on the available information.

    Args:
        location: The place of interest.
        max_driving_hours: Used to limit the surf spots based on
            the distance to `location`.
        date: Used to filter the forecast results.
        model_id: Can be any of the [litellm providers](https://docs.litellm.ai/docs/providers).

    Returns:
        A list of spot scores and reasons for the value.
    """
    max_driving_meters = driving_hours_to_meters(max_driving_hours)
    lat, lon = get_area_lat_lon(location)

    logger.info(f"Getting surfing spots around {location}")
    surf_spots = get_surfing_spots(lat, lon, max_driving_meters)

    if not surf_spots:
        logger.warning("No surfing spots found around {location}")
        return None

    spots_scores = []
    for spot_name, (spot_lat, spot_lon) in surf_spots:
        logger.info(f"Processing {spot_name}")
        logger.debug("Getting wave forecast...")
        wave_forecast = get_wave_forecast(spot_lat, spot_lon, date)
        logger.debug("Getting wind forecast...")
        wind_forecast = get_wind_forecast(spot_lat, spot_lon, date)

        logger.debug("Searching web for spot information")
        search_result = search_web(f"surf-forecast.com spot info {spot_name}")
        match = re.search(spot_info_pattern, search_result)
        if match:
            extracted_url = match.group(1)
            logger.debug(f"Visiting {extracted_url}")
            spot_info = visit_webpage(extracted_url)
        else:
            logger.debug(f"Couldn't find spot info for {spot_name}")
            continue

        logger.debug("Scoring conditions with LLM")
        response = completion(
            model="openai/gpt-4o-mini",
            messages=[
                {
                    "content": "Given the wind and wave forecast along with the spot information, "
                    "rate from 1 to 5 the expected surfing conditions."
                    f"Wind forecast:\n{wind_forecast}\n"
                    f"Wave forecast:\n{wave_forecast}\n"
                    f"Spot Information:\n{spot_info}",
                    "role": "user",
                }
            ],
            response_format=SpotScore,
        )
        spot_score = SpotScore.model_validate_json(response.choices[0].message.content)
        logger.debug(spot_score)
        spots_scores.append(spot_score)

    return spots_scores

Tools

surf_spot_finder.tools.openmeteo

get_wave_forecast(lat, lon, date)

Get wave forecast for given location.

Forecast will include:

• wave_direction (degrees)
• wave_height (meters)
• wave_period (seconds)
• sea_level_height_msl (meters)

Parameters:

Name	Type	Description	Default
lat	float	Latitude of the location.	required
lon	float	Longitude of the location.	required
date	str	Date to filter by in any valid ISO 8601 format.	required

Returns:

Type	Description
list[dict]	Hourly data for wave forecast. Example output: [ {'time': '2025-03-19T09:00', 'winddirection_10m': 140, 'windspeed_10m': 24.5}, {'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1}, {'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1}, {'time': '2025-03-19T11:00', 'winddirection_10m': 141, 'windspeed_10m': 29.2} ]

Source code in src/surf_spot_finder/tools/openmeteo.py

def get_wave_forecast(lat: float, lon: float, date: str) -> list[dict]:
    """Get wave forecast for given location.

    Forecast will include:

    - wave_direction (degrees)
    - wave_height (meters)
    - wave_period (seconds)
    - sea_level_height_msl (meters)

    Args:
        lat: Latitude of the location.
        lon: Longitude of the location.
        date: Date to filter by in any valid ISO 8601 format.

    Returns:
        Hourly data for wave forecast.
            Example output:

            ```json
            [
                {'time': '2025-03-19T09:00', 'winddirection_10m': 140, 'windspeed_10m': 24.5}, {'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1},
                {'time': '2025-03-19T10:00', 'winddirection_10m': 140, 'windspeed_10m': 27.1}, {'time': '2025-03-19T11:00', 'winddirection_10m': 141, 'windspeed_10m': 29.2}
            ]
            ```
    """
    url = "https://marine-api.open-meteo.com/v1/marine"
    params = {
        "latitude": lat,
        "longitude": lon,
        "hourly": [
            "wave_direction",
            "wave_height",
            "wave_period",
            "sea_level_height_msl",
        ],
    }
    response = requests.get(url, params=params)
    response.raise_for_status()
    data = json.loads(response.content.decode())
    hourly_data = _extract_hourly_data(data)
    if date is not None:
        date = datetime.fromisoformat(date)
        hourly_data = _filter_by_date(date, hourly_data)
    if len(hourly_data) == 0:
        raise ValueError("No data found for the given date")
    return hourly_data

get_wind_forecast(lat, lon, date)

Get wind forecast for given location.

Forecast will include:

• wind_direction (degrees)
• wind_speed (meters per second)

Parameters:

Name	Type	Description	Default
lat	float	Latitude of the location.	required
lon	float	Longitude of the location.	required
date	str	Date to filter by in any valid ISO 8601 format.	required

Returns:

Type	Description
list[dict]	Hourly data for wind forecast. Example output: [ {"time": "2025-03-18T22:00", "wind_direction": 196, "wind_speed": 9.6}, {"time": "2025-03-18T23:00", "wind_direction": 183, "wind_speed": 7.9}, ]

Source code in src/surf_spot_finder/tools/openmeteo.py

def get_wind_forecast(lat: float, lon: float, date: str) -> list[dict]:
    """Get wind forecast for given location.

    Forecast will include:

    - wind_direction (degrees)
    - wind_speed (meters per second)

    Args:
        lat: Latitude of the location.
        lon: Longitude of the location.
        date: Date to filter by in any valid ISO 8601 format.

    Returns:
        Hourly data for wind forecast.
            Example output:

            ```json
            [
                {"time": "2025-03-18T22:00", "wind_direction": 196, "wind_speed": 9.6},
                {"time": "2025-03-18T23:00", "wind_direction": 183, "wind_speed": 7.9},
            ]
            ```
    """
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": lat,
        "longitude": lon,
        "hourly": ["winddirection_10m", "windspeed_10m"],
    }
    response = requests.get(url, params=params)
    response.raise_for_status()
    data = json.loads(response.content.decode())
    hourly_data = _extract_hourly_data(data)
    date = datetime.fromisoformat(date)
    hourly_data = _filter_by_date(date, hourly_data)
    if len(hourly_data) == 0:
        raise ValueError("No data found for the given date")
    return hourly_data

surf_spot_finder.tools.openstreetmap

driving_hours_to_meters(driving_hours)

Convert driving hours to meters assuming a 70 km/h average speed.

Parameters:

Name	Type	Description	Default
driving_hours	int	The driving hours.	required

Returns:

Type	Description
int	The distance in meters.

Source code in src/surf_spot_finder/tools/openstreetmap.py

def driving_hours_to_meters(driving_hours: int) -> int:
    """Convert driving hours to meters assuming a 70 km/h average speed.


    Args:
        driving_hours: The driving hours.

    Returns:
        The distance in meters.
    """
    return driving_hours * 70 * 1000

get_area_lat_lon(area_name)

Get the latitude and longitude of an area from Nominatim.

Uses the Nominatim API.

Parameters:

Name	Type	Description	Default
area_name	str	The name of the area.	required

Returns:

Type	Description
tuple[float, float]	The area found.

Source code in src/surf_spot_finder/tools/openstreetmap.py

def get_area_lat_lon(area_name: str) -> tuple[float, float]:
    """Get the latitude and longitude of an area from Nominatim.

    Uses the [Nominatim API](https://nominatim.org/release-docs/develop/api/Search/).

    Args:
        area_name: The name of the area.

    Returns:
        The area found.
    """
    response = requests.get(
        f"https://nominatim.openstreetmap.org/search?q={area_name}&format=jsonv2",
        headers={"User-Agent": "Mozilla/5.0"},
    )
    response.raise_for_status()
    area = json.loads(response.content.decode())
    return area[0]["lat"], area[0]["lon"]

get_lat_lon_center(bounds)

Get the latitude and longitude of the center of a bounding box.

Parameters:

Name	Type	Description	Default
bounds	dict	The bounding box. { "minlat": float, "minlon": float, "maxlat": float, "maxlon": float, }	required

Returns:

Type	Description
tuple[float, float]	The latitude and longitude of the center.

Source code in src/surf_spot_finder/tools/openstreetmap.py

def get_lat_lon_center(bounds: dict) -> tuple[float, float]:
    """Get the latitude and longitude of the center of a bounding box.

    Args:
        bounds: The bounding box.

            ```json
            {
                "minlat": float,
                "minlon": float,
                "maxlat": float,
                "maxlon": float,
            }
            ```

    Returns:
        The latitude and longitude of the center.
    """
    return (
        (bounds["minlat"] + bounds["maxlat"]) / 2,
        (bounds["minlon"] + bounds["maxlon"]) / 2,
    )

get_surfing_spots(lat, lon, radius)

Get surfing spots around a given latitude and longitude.

Uses the Overpass API.

Parameters:

Name	Type	Description	Default
lat	float	The latitude.	required
lon	float	The longitude.	required
radius	int	The radius in meters.	required

Returns:

Type	Description
list[tuple[str, tuple[float, float]]]	The surfing places found.

Source code in src/surf_spot_finder/tools/openstreetmap.py

def get_surfing_spots(
    lat: float, lon: float, radius: int
) -> list[tuple[str, tuple[float, float]]]:
    """Get surfing spots around a given latitude and longitude.

    Uses the [Overpass API](https://wiki.openstreetmap.org/wiki/Overpass_API).

    Args:
        lat: The latitude.
        lon: The longitude.
        radius: The radius in meters.

    Returns:
        The surfing places found.
    """
    overpass_url = "https://overpass-api.de/api/interpreter"
    query = "[out:json];("
    query += f'nwr["natural"="beach"](around:{radius},{lat},{lon});'
    query += f'nwr["natural"="reef"](around:{radius},{lat},{lon});'
    query += ");out body geom;"
    params = {"data": query}
    response = requests.get(
        overpass_url, params=params, headers={"User-Agent": "Mozilla/5.0"}
    )
    response.raise_for_status()
    elements = response.json()["elements"]
    return [
        (element.get("tags", {}).get("name", ""), get_lat_lon_center(element["bounds"]))
        for element in elements
        if "surfing" in element.get("tags", {}).get("sport", "") and "bounds" in element
    ]