get_top_species

Instructions

Implementation Reference

• atmospore_mcp/server.py:172-188 (handler)

  The get_top_species tool handler function. An async function decorated with @mcp.tool that calls client.pollen_top(lat, lon, limit) and returns a ranked list of species with display_name, category, max_value, units, and risk_level. Wrapped in _safe_call for error handling.

  @mcp.tool(description=GET_TOP_SPECIES_DESCRIPTION)
  async def get_top_species(lat: float, lon: float, limit: int = 5) -> dict[str, Any]:
      async def call() -> Any:
          top = await client.pollen_top(lat=lat, lon=lon, limit=limit)
          return [
              {
                  "species": s.species,
                  "display_name": s.display_name,
                  "category": s.category,
                  "max_value": s.max_value,
                  "units": s.units,
                  "risk_level": s.risk_level,
              }
              for s in top
          ]
  
      return await _safe_call(call())

• atmospore_mcp/server.py:36-46 (schema)

  GET_TOP_SPECIES_DESCRIPTION — documentation string describing input params (lat, lon, limit=5) and output shape (species, display_name, max_value, risk_level, category). Used by the LLM during MCP handshake.

  GET_TOP_SPECIES_DESCRIPTION = """Get the top contributing pollen species at a specific point today.
  
  Use this to answer "what pollen is highest in Oslo right now?" or "which trees are blooming in
  Bergen?". Returns a ranked list (highest first). Each species includes:
  - species (slug, e.g. 'birch')
  - display_name (human-readable, e.g. 'Birch')
  - max_value in grains/m³
  - risk_level
  - category ('tree' | 'grass' | 'weed')
  
  `limit` (default 5) caps the list length."""

• atmospore_mcp/server.py:172-172 (registration)

  Tool registration via @mcp.tool(description=GET_TOP_SPECIES_DESCRIPTION) decorator on the get_top_species function inside build_server().

  @mcp.tool(description=GET_TOP_SPECIES_DESCRIPTION)

• atmospore_mcp/server.py:117-147 (helper)

  _safe_call helper — wraps async client calls to catch AuthenticationError, RateLimitError, APIError, and unexpected exceptions, returning structured dicts with ok/error/message for the LLM.

  async def _safe_call(coro) -> dict[str, Any]:
      """Wrap a client call so structured errors reach the LLM cleanly."""
      try:
          return {"ok": True, "data": await coro}
      except AuthenticationError as e:
          return {
              "ok": False,
              "error": "authentication_failed",
              "message": str(e),
              "hint": "Check ATMOSPORE_API_KEY. Get a free key at https://atmospore.com/account.",
          }
      except RateLimitError as e:
          return {
              "ok": False,
              "error": "quota_exceeded",
              "message": str(e),
              "limit": e.limit,
              "used": e.used,
              "resets_at": e.resets_at,
              "hint": "Daily quota hit. Upgrade at https://atmospore.com/plans for higher limits.",
          }
      except APIError as e:
          return {
              "ok": False,
              "error": "api_error",
              "status": e.status,
              "message": str(e),
          }
      except Exception as e:  # noqa: BLE001 — surface unknown errors to LLM
          logger.exception("Unexpected error in MCP tool")
          return {"ok": False, "error": "unexpected_error", "message": str(e)}