Source code for tooluniverse.inaturalist_tool
# inaturalist_tool.py
"""
iNaturalist API tool for ToolUniverse.
iNaturalist is a citizen science platform for biodiversity observations,
connecting millions of people worldwide to document and identify species.
The API provides access to taxa, observations, and species counts.
API: https://api.inaturalist.org/v1/
No authentication required. Free public access.
Rate limit: Be respectful (no more than ~1 req/sec).
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
INAT_BASE_URL = "https://api.inaturalist.org/v1"
# iNaturalist returns up to 100 place listings per taxon out of a count that
# can be far larger (1,588 for the monarch butterfly), and up to 29 regional
# conservation statuses. Both are capped here with the cap and the upstream
# count reported, rather than reduced to their first element.
STATUSES_PER_TAXON = 25
PLACE_LISTINGS_PER_TAXON = 25
WIKIPEDIA_SUMMARY_CHARACTERS = 600
PHOTOS_PER_OBSERVATION = 10
def _conservation_status(status: dict) -> dict:
"""One conservation status, with the authority that issued it.
The row used to be `cs.get("status")`: for Danaus plexippus that is the
bare string "G4", which is a NatureServe rank and not an IUCN category,
and nothing said which. A status without its authority and its place is
not a status.
"""
if not isinstance(status, dict):
return {}
place = status.get("place") or {}
return {
"status": status.get("status"),
"authority": status.get("authority"),
"iucn_code": status.get("iucn"),
"description": status.get("description"),
"url": status.get("url"),
# null for a global status; a place id and name for a regional one.
"place_id": place.get("id") if isinstance(place, dict) else None,
"place_name": place.get("display_name") or place.get("name")
if isinstance(place, dict)
else None,
}
def _conservation_statuses(record: dict) -> list:
"""The regional statuses beside the headline one. The monarch has 29,
from the Canadian Species at Risk Act to NatureServe state ranks, and
the row reported one."""
statuses = record.get("conservation_statuses") or []
rows = [_conservation_status(x) for x in statuses if isinstance(x, dict)]
return [r for r in rows if r.get("status")][:STATUSES_PER_TAXON]
def _place_listings(record: dict) -> list:
"""Where the taxon is listed and whether it is native or introduced.
`establishment_means` is "introduced" for the monarch in Australia and
Guam and "native" elsewhere, which for a taxonomy tool is a fact about
the species, not presentation.
"""
out = []
for listing in record.get("listed_taxa") or []:
if not isinstance(listing, dict):
continue
place = listing.get("place") or {}
listed_in = listing.get("list") or {}
out.append(
{
"place_id": place.get("id") if isinstance(place, dict) else None,
"place_name": place.get("display_name") or place.get("name")
if isinstance(place, dict)
else None,
"establishment_means": listing.get("establishment_means"),
"list_title": listed_in.get("title")
if isinstance(listed_in, dict)
else None,
}
)
if len(out) >= PLACE_LISTINGS_PER_TAXON:
break
return out
def _photos(record: dict) -> list:
"""The observation's photos with the terms they may be used under.
iNaturalist photos carry their own copyright: some are all rights
reserved, some Creative Commons. A bare URL does not say which, so
every photo is returned with its attribution line and licence code.
``license_code`` is null for an all-rights-reserved photo.
"""
rows = []
for photo in (record.get("photos") or [])[:PHOTOS_PER_OBSERVATION]:
if not isinstance(photo, dict):
continue
rows.append(
{
"id": photo.get("id"),
"url": photo.get("url"),
"attribution": photo.get("attribution"),
"license_code": photo.get("license_code"),
}
)
return rows
[docs]
@register_tool("INaturalistTool")
class INaturalistTool(BaseTool):
"""
Tool for querying iNaturalist biodiversity data.
iNaturalist aggregates citizen science observations from around the world.
Research-grade observations are community-verified and used in scientific
research. Covers all kingdoms of life with over 150 million observations.
Supports: taxa search, taxon details, observation search, species counts.
No authentication required.
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.timeout = tool_config.get("timeout", 30)
fields = tool_config.get("fields", {})
self.endpoint = fields.get("endpoint", "search_taxa")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the iNaturalist API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"iNaturalist API request timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {"status": "error", "error": "Failed to connect to iNaturalist API"}
except requests.exceptions.HTTPError as e:
return {
"status": "error",
"error": f"iNaturalist API HTTP error: {e.response.status_code}",
}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying iNaturalist: {str(e)}",
}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate iNaturalist endpoint."""
if self.endpoint == "search_taxa":
return self._search_taxa(arguments)
elif self.endpoint == "get_taxon":
return self._get_taxon(arguments)
elif self.endpoint == "search_observations":
return self._search_observations(arguments)
elif self.endpoint == "species_counts":
return self._get_species_counts(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
@staticmethod
def _page_metadata(
total: int, page: int, per_page: int, returned: int
) -> Dict[str, Any]:
accessible_pages = 10000 // per_page
accessible_total = min(total, accessible_pages * per_page)
has_more = bool(
returned and page < accessible_pages and page * per_page < total
)
return {
"returned": returned,
"total_results": total,
"total_results_is_exact": True,
"page": page,
"per_page": per_page,
"has_more": has_more,
"next_page": page + 1 if has_more else None,
"accessible_results": accessible_total,
"accessible_pages": min(
accessible_pages, (total + per_page - 1) // per_page
),
"upstream_truncated": total > accessible_total,
"truncated": page > 1 or has_more or total > accessible_total,
"result_set_complete": page == 1 and returned == total,
"api_result_window_limit": 10000,
"pagination_scope": "inaturalist_page_10000_result_window",
"warning": (
"iNaturalist matched more records than this endpoint's page-based "
"10,000-result window exposes. Refine the query."
if total > accessible_total
else None
),
}
[docs]
@staticmethod
def _validate_page(
arguments: Dict[str, Any], default_per_page: int, maximum: int
) -> tuple[int, int] | Dict[str, Any]:
per_page = arguments.get("per_page", default_per_page)
page = arguments.get("page", 1)
if per_page is None:
per_page = default_per_page
if page is None:
page = 1
if (
isinstance(per_page, bool)
or not isinstance(per_page, int)
or not 1 <= per_page <= maximum
):
return {
"status": "error",
"error": f"per_page must be an integer between 1 and {maximum}",
"parameter": "per_page",
}
if isinstance(page, bool) or not isinstance(page, int) or page < 1:
return {
"status": "error",
"error": "page must be an integer greater than or equal to 1",
"parameter": "page",
}
if page * per_page > 10000:
return {
"status": "error",
"error": "iNaturalist requires page * per_page <= 10000",
"parameter": "page",
}
return page, per_page
[docs]
def _search_taxa(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search for taxa by name."""
query = arguments.get("query", "")
if not query:
return {"status": "error", "error": "query parameter is required"}
pagination = self._validate_page(arguments, 10, 200)
if isinstance(pagination, dict):
return pagination
page, per_page = pagination
url = f"{INAT_BASE_URL}/taxa"
params = {"q": query, "per_page": per_page, "page": page}
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
results = []
for r in data.get("results", []):
cs = r.get("conservation_status")
results.append(
{
"id": r.get("id"),
"name": r.get("name", ""),
"common_name": r.get("preferred_common_name"),
"rank": r.get("rank"),
"observations_count": r.get("observations_count", 0),
"is_active": r.get("is_active", True),
"conservation_status": cs.get("status") if cs else None,
# The same status with the authority that issued it, and
# the regional ones beside it.
"conservation_status_detail": _conservation_status(cs),
"conservation_statuses": _conservation_statuses(r),
"conservation_status_count": len(
r.get("conservation_statuses") or []
),
"extinct": r.get("extinct"),
"iconic_taxon_name": r.get("iconic_taxon_name"),
"iconic_taxon_id": r.get("iconic_taxon_id"),
# Two taxa can share a common name across kingdoms
# ("Monarch" the butterfly, "Monarch Flycatchers" the
# bird family); the lineage is what separates them.
# This endpoint sends the path as ids only, each one a
# taxon id accepted by iNaturalist_get_taxon.
"ancestor_ids": r.get("ancestor_ids")
or [
int(part)
for part in str(r.get("ancestry") or "").split("/")
if part.isdigit()
],
"wikipedia_url": r.get("wikipedia_url"),
}
)
return {
"status": "success",
"data": results,
"metadata": {
"source": "iNaturalist",
"query": query,
**self._page_metadata(
data.get("total_results", len(results)),
page,
params["per_page"],
len(results),
),
},
}
[docs]
def _get_taxon(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get detailed taxon information by ID."""
taxon_id = arguments.get("taxon_id")
if taxon_id is None:
return {"status": "error", "error": "taxon_id parameter is required"}
url = f"{INAT_BASE_URL}/taxa/{taxon_id}"
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
data = response.json()
results = data.get("results", [])
if not results:
return {"status": "error", "error": f"Taxon ID {taxon_id} not found"}
r = results[0]
cs = r.get("conservation_status")
summary = (r.get("wikipedia_summary") or "").strip()
summary_truncated = len(summary) > WIKIPEDIA_SUMMARY_CHARACTERS
if summary_truncated:
summary = summary[:WIKIPEDIA_SUMMARY_CHARACTERS].rstrip() + "..."
summary = summary or None
# Build ancestry
ancestors = r.get("ancestors", [])
ancestry = []
for a in ancestors:
ancestry.append(
{
"name": a.get("name", ""),
"rank": a.get("rank"),
"id": a.get("id"),
}
)
return {
"data": {
"id": r.get("id"),
"name": r.get("name", ""),
"common_name": r.get("preferred_common_name"),
"rank": r.get("rank"),
"observations_count": r.get("observations_count", 0),
"is_active": r.get("is_active", True),
"conservation_status": cs.get("status") if cs else None,
"conservation_status_detail": _conservation_status(cs),
"conservation_statuses": _conservation_statuses(r),
"conservation_status_count": len(
r.get("conservation_statuses") or []
),
"extinct": r.get("extinct"),
"iconic_taxon_name": r.get("iconic_taxon_name"),
"iconic_taxon_id": r.get("iconic_taxon_id"),
"place_listings": _place_listings(r),
"place_listings_returned": len(_place_listings(r)),
"place_listings_in_response": len(r.get("listed_taxa") or []),
"place_listings_total": r.get("listed_taxa_count"),
"wikipedia_summary": summary,
"wikipedia_summary_truncated": summary_truncated,
"wikipedia_url": r.get("wikipedia_url"),
"ancestry": ancestry,
},
"metadata": {
"source": "iNaturalist",
"statuses_per_taxon": STATUSES_PER_TAXON,
"place_listings_per_taxon": PLACE_LISTINGS_PER_TAXON,
"listings_note": "iNaturalist returns at most 100 place "
"listings per taxon however many it holds "
"(place_listings_total); place_listings is the first "
"place_listings_per_taxon of those returned.",
},
}
[docs]
def _search_observations(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search for species observations."""
pagination = self._validate_page(arguments, 10, 200)
if isinstance(pagination, dict):
return pagination
page, per_page = pagination
params = {}
if arguments.get("taxon_id"):
params["taxon_id"] = arguments["taxon_id"]
if arguments.get("query"):
params["taxon_name"] = arguments["query"]
if arguments.get("place_id"):
params["place_id"] = arguments["place_id"]
quality_grade = arguments.get("quality_grade") or "research"
if quality_grade not in {"research", "needs_id", "casual"}:
return {
"status": "error",
"error": "quality_grade must be one of: research, needs_id, casual",
"parameter": "quality_grade",
}
id_below = arguments.get("id_below")
if id_below is not None and (
isinstance(id_below, bool) or not isinstance(id_below, int) or id_below < 1
):
return {
"status": "error",
"error": "id_below must be a positive integer",
"parameter": "id_below",
}
if id_below is not None and page != 1:
return {
"status": "error",
"error": "page must be 1 when id_below cursor pagination is used",
"parameter": "page",
}
params["quality_grade"] = quality_grade
params["per_page"] = per_page
params["page"] = page
params["order_by"] = "id"
params["order"] = "desc"
if id_below is not None:
params["id_below"] = id_below
if not params.get("taxon_id") and not params.get("taxon_name"):
return {"status": "error", "error": "Either taxon_id or query is required"}
url = f"{INAT_BASE_URL}/observations"
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
results = []
for obs in data.get("results", []):
taxon = obs.get("taxon") or {}
geojson = obs.get("geojson") or {}
coords = geojson.get("coordinates", [None, None])
photos = obs.get("photos", [])
results.append(
{
"id": obs.get("id"),
"species_name": taxon.get("name"),
"common_name": taxon.get("preferred_common_name"),
"observed_on": obs.get("observed_on"),
"place_guess": obs.get("place_guess"),
"latitude": coords[1] if len(coords) > 1 else None,
"longitude": coords[0] if coords else None,
"quality_grade": obs.get("quality_grade"),
"user": obs.get("user", {}).get("login"),
"photo_url": photos[0].get("url") if photos else None,
# photo_url alone says nothing about whether the photo
# may be reused; photos carries each one's attribution
# line and licence code.
"photos": _photos(obs),
"photo_count": len(photos),
}
)
total = data.get("total_results", len(results))
total_is_exact = isinstance(total, int) and not isinstance(total, bool)
returned = len(results)
page_has_more = bool(
total_is_exact
and returned
and page * per_page < total
and (page + 1) * per_page <= 10000
)
cursor_has_more = bool(total_is_exact and returned and page * per_page < total)
next_id_below = results[-1]["id"] if cursor_has_more else None
return {
"status": "success",
"data": results,
"metadata": {
"source": "iNaturalist",
"returned": returned,
"total_results": total if total_is_exact else None,
"total_results_is_exact": total_is_exact,
"total_results_scope": (
"records_below_id_below" if id_below is not None else "all_matches"
),
"page": page,
"per_page": per_page,
"page_has_more": page_has_more,
"next_page": page + 1 if page_has_more else None,
"has_more": cursor_has_more,
"id_below": id_below,
"next_id_below": next_id_below,
"truncated": page > 1 or cursor_has_more or id_below is not None,
"result_set_complete": bool(
total_is_exact
and id_below is None
and page == 1
and returned == total
),
"api_result_window_limit": 10000,
"pagination_scope": "inaturalist_id_below_sliding_window",
"warning": (
"Page-number pagination is limited to 10,000 results. Continue "
"with page=1 and metadata.next_id_below to traverse beyond it."
if total_is_exact and total > 10000
else None
),
},
}
[docs]
def _get_species_counts(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get species counts for a taxon group or location."""
pagination = self._validate_page(arguments, 20, 500)
if isinstance(pagination, dict):
return pagination
page, per_page = pagination
params = {}
if arguments.get("taxon_id"):
params["taxon_id"] = arguments["taxon_id"]
if arguments.get("place_id"):
params["place_id"] = arguments["place_id"]
quality_grade = arguments.get("quality_grade") or "research"
if quality_grade not in {"research", "needs_id", "casual"}:
return {
"status": "error",
"error": "quality_grade must be one of: research, needs_id, casual",
"parameter": "quality_grade",
}
params["quality_grade"] = quality_grade
params["per_page"] = per_page
params["page"] = page
if not params.get("taxon_id") and not params.get("place_id"):
return {
"status": "error",
"error": "Either taxon_id or place_id is required",
}
url = f"{INAT_BASE_URL}/observations/species_counts"
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
results = []
for r in data.get("results", []):
taxon = r.get("taxon", {})
results.append(
{
"taxon_id": taxon.get("id"),
"name": taxon.get("name", ""),
"common_name": taxon.get("preferred_common_name"),
"rank": taxon.get("rank"),
"count": r.get("count", 0),
}
)
return {
"status": "success",
"data": results,
"metadata": {
"source": "iNaturalist",
"total_species": data.get("total_results", len(results)),
**self._page_metadata(
data.get("total_results", len(results)),
params["page"],
params["per_page"],
len(results),
),
},
}