Fermi LAT Data Query API
The Fermi LAT Data Query API provides a programmatic interface for submitting,
monitoring, and retrieving Fermi LAT data products. It is designed for scripted
and automated access using standard HTTP tools such as curl, Python,
or workflow systems.
curl and Python. The Python snippets use
requests
(pip install requests). Every example assumes:
import requests
from requests.adapters import HTTPAdapter, Retry
from urllib.parse import urljoin
BASE = "https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/"
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(
total=5, backoff_factor=1, status_forcelist=[502, 503, 504])))
curl never sees this because each
invocation opens a fresh connection, but a reused
requests.Session can hit the stale socket and raise
ConnectionError: RemoteDisconnected — typically on the
second request of a polling loop. The retry adapter reopens the connection
transparently. Retries apply only to GET requests (the urllib3 default), so
a POST can never submit the same query twice.
Typical Workflow
POST /query -> receive query_id
GET /query/{id}/status -> poll until complete
GET /query/{id}/results -> retrieve file metadata and download URLs
-> download each file from the url field returned by /results
API Architecture
The API acts as a frontend to the existing Fermi LAT backend infrastructure:
- Client submits a query via HTTP (JSON payload)
- The API validates and normalizes parameters
- A query is submitted to the Queue Manager
-
One or more backend servers process the request:
- Photon Server
- Spacecraft Server
- Event Server (if applicable)
- Results are staged on the HEASARC FTP area
- The API exposes status and result metadata
Base URL
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1
All endpoints below are relative to this base URL.
1. Submit a Query
POST /query
Submits a new Fermi LAT data query for processing.
Request Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
coordfield |
string | Yes | Coordinates as "RA,Dec" or target name |
shapefield |
number | Yes | Search radius in degrees |
coordsystem |
string | Yes | J2000, B1950, or Galactic |
timefield |
string | Yes | Start and stop time as "t1,t2" |
timetype |
string | Yes | Gregorian, MJD, or MET |
energyfield |
string | Yes | Energy range in MeV as "min,max" |
photonOrExtendedOrNone |
string | No | Photon, Extended, or None |
spacecraft |
string | No | Include spacecraft data: "on" |
zenithangle |
number | No | Max zenith angle in degrees (default: 180, no cut applied) |
destination |
string | No | Backend query destination. query is the only accepted
value and the default when omitted; any other value is rejected with
HTTP 400 |
query_id |
string | No | Existing query ID, when re-submitting |
Example Request
curl -k -X POST \
-H "Content-Type: application/json" \
-d '{
"coordfield": "128.836,-45.1764",
"shapefield": 15,
"coordsystem": "J2000",
"timefield": "772109936,787661936",
"timetype": "MET",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on"
}' \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query
Python
query = {
"coordfield": "128.836,-45.1764",
"shapefield": 15,
"coordsystem": "J2000",
"timefield": "772109936,787661936",
"timetype": "MET",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on",
}
r = session.post(urljoin(BASE, "query"), json=query, timeout=60)
if r.status_code == 400:
raise SystemExit(r.json()["error"])
r.raise_for_status()
query_id = r.json()["query_id"]
print(query_id)
Example Response
{
"query_id": "L2601082002167F48EE3069",
"status": "submitted",
"status_url": "/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/status",
"results_url": "/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/results"
}
query_id — it is required
for all subsequent requests. The status_url and
results_url fields are server-relative paths; prefix them with the
scheme and host to request them.
Example: Submit with Zenith Angle
curl -k -X POST \
-H "Content-Type: application/json" \
-d '{
"coordfield": "128.836,-45.1764",
"shapefield": 15,
"coordsystem": "J2000",
"timefield": "772109936,787661936",
"timetype": "MET",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on",
"zenithangle": 85
}' \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query
Python
query["zenithangle"] = 85
r = session.post(urljoin(BASE, "query"), json=query, timeout=60)
r.raise_for_status()
query_id = r.json()["query_id"]
2. Check Query Status
GET /query/{query_id}/status
Returns the current state of the query, including queue position and server execution status.
Example Request
curl -k \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/status
Python
r = session.get(urljoin(BASE, f"query/{query_id}/status"), timeout=60)
r.raise_for_status()
print(r.json()["state"])
Example Response
{
"query_id": "L2601082002167F48EE3069",
"queue_status": [
{ "server": "Photon Server", "queue_rank": 0, "time_remaining": "Unknown" },
{ "server": "Spacecraft Server", "queue_rank": 0, "time_remaining": "Unknown" }
],
"running_status": [
{ "server": "Photon Server", "status": "Completed", "time_remaining": "N/A" }
],
"servers_status": [
{ "name": "Photon Server", "position": "Query complete", "time_remaining": "N/A" }
],
"state": "Query completed"
}
Python: poll until the query finishes
import time
def wait_for(query_id, every=15, timeout=3600):
"""Poll the status endpoint until the backend reports the query is done."""
url = urljoin(BASE, f"query/{query_id}/status")
deadline = time.monotonic() + timeout
while True:
r = session.get(url, timeout=60)
r.raise_for_status()
state = str(r.json().get("state", ""))
print(state)
if "complet" in state.lower():
return
if any(w in state.lower() for w in ("fail", "error", "abort")):
raise SystemExit(f"Query did not complete: {state}")
if time.monotonic() > deadline:
raise SystemExit(f"Timed out; last state was: {state}")
time.sleep(every)
wait_for(query_id)
3. List Result Files
GET /query/{query_id}/results
Returns metadata for all result files associated with the query, including a
ready-to-use download url for each one.
Example Request
curl -k \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/results
Python
r = session.get(urljoin(BASE, f"query/{query_id}/results"), timeout=60)
r.raise_for_status()
files = r.json()["files"]
for f in files:
print(f["name"], f["size"], "MB", f["status"])
Example Response
{
"query_id": "L2601082002167F48EE3069",
"state": 2,
"files": [
{
"name": "L2601082002167F48EE3069_PH00.fits",
"entries": 703,
"size": "0.09",
"status": "available",
"url": "https://fermi.gsfc.nasa.gov/FTP/fermi/data/lat/test/queries/L2601082002167F48EE3069_PH00.fits"
}
]
}
Listing Only File Names
curl -k \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/results \
| jq -r '.files[].name'
Python
names = [f["name"] for f in files]
4. Download Result Files
Result files are staged on the HEASARC FTP infrastructure. Each entry returned
by /results includes a complete url — use that
rather than assembling the path yourself, so downloads keep working if the
storage location changes.
Download Every File From a Query
curl -k \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/results \
| jq -r '.files[].url' \
| xargs -n1 curl -L -O
Python: download every file
from pathlib import Path
outdir = Path("lat_data")
outdir.mkdir(exist_ok=True)
for f in files:
if f["status"] != "available":
continue
dest = outdir / f["name"]
with session.get(f["url"], stream=True, timeout=300) as r:
r.raise_for_status()
with open(dest, "wb") as fh:
for chunk in r.iter_content(chunk_size=1 << 20):
fh.write(chunk)
print(dest)
Download a Single File
curl -L -O \
"$(curl -k https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query/L2601082002167F48EE3069/results \
| jq -r '.files[0].url')"
Python: download a single file
f = files[0]
with session.get(f["url"], stream=True, timeout=300) as r:
r.raise_for_status()
with open(f["name"], "wb") as fh:
for chunk in r.iter_content(chunk_size=1 << 20):
fh.write(chunk)
url field. Each entry from
/results carries a complete download URL. Building the FTP path by
hand works until the storage location moves; the returned URL keeps working.
All-Sky Queries
All-sky queries retrieve data covering the entire sky and are treated specially by the backend. The following rules apply:
shapefieldmust be > 60° (typically 180)- Observation window must be <= 24 hours
- Coordinates may be set to
0.0,0.0or left blank
Example: All-Sky Query
curl -k -X POST \
-H "Content-Type: application/json" \
-d '{
"coordfield": "0.0,0.0",
"shapefield": 180,
"coordsystem": "J2000",
"timefield": "2008-08-04 15:43:36,2008-08-05 09:14:33",
"timetype": "Gregorian",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on"
}' \
https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/query
Python
allsky = {
"coordfield": "0.0,0.0",
"shapefield": 180, # must be > 60
"coordsystem": "J2000",
"timefield": "2008-08-04 15:43:36,2008-08-05 09:14:33", # <= 24 hours
"timetype": "Gregorian",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on",
}
r = session.post(urljoin(BASE, "query"), json=allsky, timeout=60)
r.raise_for_status()
Error Responses
Errors are returned as JSON with an error field and an appropriate
HTTP status code. Some parameter-validation failures also include an
error_type field categorizing the error; rely on
error, which is always present.
{
"error": "Zenith angle must be between 0 and 180 degrees.",
"error_type": "Zenith Angle Error"
}
{ "error": "Query not found" }
| Status | Meaning |
|---|---|
400 |
Invalid request: bad JSON, unknown parameter, or failed validation |
403 |
Requesting address is blocked |
404 |
Query ID not found |
500 |
Server-side failure while preparing or submitting the query |
Python: reading the error
r = session.post(urljoin(BASE, "query"), json=query, timeout=60)
if not r.ok:
try:
message = r.json()["error"]
except ValueError:
message = r.text
raise SystemExit(f"HTTP {r.status_code}: {message}")
Complete Python Example
Submit a query, wait for the backend, and download every result file. Requires
requests.
#!/usr/bin/env python3
"""Submit a Fermi LAT data query, wait for it, and download the results."""
import time
from pathlib import Path
from urllib.parse import urljoin
import requests
from requests.adapters import HTTPAdapter, Retry
BASE = "https://fermi.gsfc.nasa.gov/ssc/data/access/lat/query/api/v1/"
session = requests.Session()
# The server closes idle keep-alive connections; retry GETs transparently.
# POSTs are never retried, so a query cannot be submitted twice.
session.mount("https://", HTTPAdapter(max_retries=Retry(
total=5, backoff_factor=1, status_forcelist=[502, 503, 504])))
QUERY = {
"coordfield": "128.836,-45.1764",
"shapefield": 15,
"coordsystem": "J2000",
"timefield": "772109936,787661936",
"timetype": "MET",
"energyfield": "100,300000",
"photonOrExtendedOrNone": "Photon",
"spacecraft": "on",
}
# 1. submit
r = session.post(urljoin(BASE, "query"), json=QUERY, timeout=60)
if r.status_code == 400:
raise SystemExit(r.json()["error"])
r.raise_for_status()
query_id = r.json()["query_id"]
print("query_id:", query_id)
# 2. poll
status_url = urljoin(BASE, f"query/{query_id}/status")
while True:
r = session.get(status_url, timeout=60)
r.raise_for_status()
state = str(r.json().get("state", ""))
print(" ", state)
if "complet" in state.lower():
break
if any(w in state.lower() for w in ("fail", "error", "abort")):
raise SystemExit(f"Query did not complete: {state}")
time.sleep(15)
# 3. list
r = session.get(urljoin(BASE, f"query/{query_id}/results"), timeout=60)
r.raise_for_status()
files = r.json()["files"]
# 4. download
outdir = Path("lat_data")
outdir.mkdir(exist_ok=True)
for f in files:
if f["status"] != "available":
continue
dest = outdir / f["name"]
with session.get(f["url"], stream=True, timeout=300) as resp:
resp.raise_for_status()
with open(dest, "wb") as fh:
for chunk in resp.iter_content(chunk_size=1 << 20):
fh.write(chunk)
print(" ", dest)
from astropy.io import fits
with fits.open("lat_data/L2601082002167F48EE3069_PH00.fits") as hdul:
hdul.info()
events = hdul["EVENTS"].data