get_pollen

Instructions

Implementation Reference

• atmospore_mcp/server.py:149-170 (handler)

  The main handler function for the get_pollen tool. It takes lat, lon, and optional forecast_days, calls the AtmosporeClient.pollen() method, and returns a dict with per-day date, overall_risk, and filtered per-species values (dropping zero-value entries). Results are wrapped via _safe_call for structured error handling.

  @mcp.tool(description=GET_POLLEN_DESCRIPTION)
  async def get_pollen(lat: float, lon: float, forecast_days: int = 1) -> dict[str, Any]:
      async def call() -> Any:
          days = await client.pollen(lat=lat, lon=lon, forecast_days=forecast_days)
          return [
              {
                  "date": d.date,
                  "overall_risk": d.overall_risk,
                  "species": {
                      slug: {
                          "value": lvl.value,
                          "units": lvl.units,
                          "risk_level": lvl.risk_level,
                      }
                      for slug, lvl in d.pollen_levels.items()
                      if lvl.value > 0  # drop zero entries to keep the LLM's context clean
                  },
              }
              for d in days
          ]
  
      return await _safe_call(call())

• atmospore_mcp/server.py:149-150 (registration)

  The tool is registered using the @mcp.tool decorator on the FastMCP instance within build_server(). The tool name is automatically derived from the function name 'get_pollen'.

  @mcp.tool(description=GET_POLLEN_DESCRIPTION)
  async def get_pollen(lat: float, lon: float, forecast_days: int = 1) -> dict[str, Any]:

• atmospore_mcp/server.py:23-23 (schema)

  The tool description/schema is defined as a constant string (GET_POLLEN_DESCRIPTION) used in the @mcp.tool decorator, describing the tool's behavior and output shape to the LLM.

  GET_POLLEN_DESCRIPTION = """Get the daily pollen forecast for a specific point on Earth.

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

  The _safe_call helper wraps all tool handler coroutines, catching known error types (AuthenticationError, RateLimitError, APIError) and returning structured error dicts so the LLM can present actionable messages to the user.

  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)}