nonprofit_fetch_nonprofit_by_ein
Retrieve a US nonprofit's financial history, revenue, expenses, assets, and IRS registration status using its EIN.
Instructions
Use this to research a US charity or nonprofit by EIN number. Provide the EIN with or without dash. Returns financial history, revenue, expenses, assets, and IRS registration status.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| ein | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- datanexus/tools/t04.py:110-208 (handler)The main handler function for the tool. Decorated with @mcp.tool(), @with_timeout, @verify_entitlement('T04'). Accepts an EIN string, normalizes it, checks cache, circuit breaker, performs lookup (Redis or live CSV), builds response payload with markdown, and caches the result. This is the actual implementation of the 'nonprofit_fetch_nonprofit_by_ein' tool.
@mcp.tool() @with_timeout @verify_entitlement("T04") async def fetch_nonprofit_by_ein(ein: str) -> dict: """Use this to research a US charity or nonprofit by EIN number. Provide the EIN with or without dash. Returns financial history, revenue, expenses, assets, and IRS registration status.""" # Normalise EIN — strip dashes, leading zeros ein_clean = ein.replace("-", "").strip() params = {"ein": ein_clean} async with AuditContext("T04", params, "1.0") as ctx: phash = make_params_hash(params) # ── 1. Cache check ──────────────────────────────────────────────────── cached = get_cached("T04", phash) if cached: ctx.set_cache_hit(True) log.info("t04.fetch_nonprofit_by_ein cache_hit ein=%s", ein_clean) return { **cached, **standard_response_fields( ctx.query_hash, cached.get("data_as_of", ""), cached.get("ingest_healthy", True), ), "cache_hit": True, } # ── 2. Circuit breaker check ────────────────────────────────────────── bmf_down = is_tripped("irs_bmf") teos_down = is_tripped("irs_teos") if bmf_down and teos_down: archive = get_cached("T04", phash + "_archive") ctx.set_error(ErrorCode.CIRCUIT_OPEN) return { "tool_id": "T04", "data": archive or {}, "markdown_output": _archive_markdown(archive, ein_clean), "staleness_notice": get_staleness_notice( "irs_bmf", (archive or {}).get("data_as_of", "unknown"), ), "disclaimer": T04_DISCLAIMER, "cache_hit": False, "sha256_hash": "", **standard_response_fields(ctx.query_hash, "", False), } # ── 3. Live lookup — Redis BMF index first ──────────────────────────── result = await _lookup_ein(ein_clean) ingest_healthy = True if not result: # BMF worker hasn't run yet or EIN not found ingest_healthy = False return error_response( error_code=ErrorCode.NOT_FOUND, message=f"EIN {ein_clean} not found in IRS EO BMF. " "Verify the EIN is correct and the organisation is active.", query_hash=ctx.query_hash, retry_after=0, ingest_healthy=False, ) # ── 4. Build payload ────────────────────────────────────────────────── raw_bytes = json.dumps(result).encode() payload_hash = compute_payload_hash(raw_bytes) markdown = _build_nonprofit_markdown(result, ein_clean) data_as_of = datetime.now(timezone.utc).isoformat() payload = { "tool_id": "T04", "source_url": "https://www.irs.gov/pub/irs-soi/eo1.csv", "fetch_timestamp": data_as_of, "cache_hit": False, "staleness_notice": None, "sha256_hash": payload_hash, "data": result, "markdown_output": markdown, "disclaimer": T04_DISCLAIMER, "data_as_of": data_as_of, "ingest_healthy": ingest_healthy, } # ── 5. Store in cache ───────────────────────────────────────────────── set_cached("T04", phash, payload, IRS_BMF_TTL) set_cached("T04", phash + "_archive", payload, IRS_BMF_TTL * 4) ctx.set_cache_hit(False) record_success_sync("irs_bmf") log.info("t04.fetch_nonprofit_by_ein ok ein=%s name=%s", ein_clean, result.get("name", "")) return { **payload, **standard_response_fields(ctx.query_hash, data_as_of, ingest_healthy), } - datanexus/tools/nonprofit.py:13-17 (registration)Registration of the tool on the 'nonprofit' sub-server FastMCP instance. The function is imported from t04.py and registered via nonprofit.tool()(fetch_nonprofit_by_ein).
nonprofit = FastMCP("DataNexus Nonprofit") nonprofit.tool()(fetch_nonprofit_by_ein) nonprofit.tool()(search_nonprofits_by_name) nonprofit.tool()(fetch_charity_uk) - datanexus/tools/meta.py:17-17 (registration)The tool is listed in TOOL_REGISTRY with its searchable task description for the search_datanexus_tools meta-tool.
{"name": "nonprofit_fetch_nonprofit_by_ein", "task": "research a US charity or nonprofit by EIN number"}, - datanexus/main.py:154-154 (registration)The 'nonprofit' sub-server (containing the tool) is mounted to the main FastMCP server with namespace 'nonprofit', making the tool accessible as 'nonprofit_fetch_nonprofit_by_ein'.
main.mount(nonprofit, namespace="nonprofit") - datanexus/tools/t04.py:452-505 (helper)Internal helper functions that perform the actual EIN lookup: _lookup_ein (checks Redis cache populated by IRSBMFWorker, falls back to live CSV) and _lookup_ein_csv_live (streams IRS EO BMF CSV files to find the EIN).
async def _lookup_ein(ein_clean: str) -> Optional[dict]: # 1. Redis cache (populated by IRSBMFWorker) from datanexus.core.cache import _get_redis # type: ignore[attr-defined] r = _get_redis() if r is not None: try: raw = r.get(f"datanexus:T04:bmf:{ein_clean}") if raw: return json.loads(raw) except Exception: pass # 2. Live CSV stream fallback (BMF worker not yet run) return await _lookup_ein_csv_live(ein_clean) async def _lookup_ein_csv_live(ein_clean: str) -> Optional[dict]: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, headers=_HTTP_HEADERS, follow_redirects=True ) as client: for url in _IRS_BMF_URLS: try: async with client.stream("GET", url) as resp: resp.raise_for_status() content = await resp.aread() text = content.decode("utf-8", errors="replace") reader = csv.DictReader(io.StringIO(text)) for row in reader: if row.get("EIN", "").strip() == ein_clean: record_success_sync("irs_bmf") return { "ein": row.get("EIN", "").strip(), "name": row.get("NAME", "").strip(), "street": row.get("STREET", "").strip(), "city": row.get("CITY", "").strip(), "state": row.get("STATE", "").strip(), "zip": row.get("ZIP", "").strip(), "ntee_code": row.get("NTEE_CD", "").strip(), "ruling": row.get("RULING", "").strip(), "subsection": row.get("SUBSECTION", "").strip(), "status": row.get("STATUS", "").strip(), "tax_period": row.get("TAX_PERIOD", "").strip(), "asset_amt": row.get("ASSET_AMT", "").strip(), "income_amt": row.get("INCOME_AMT", "").strip(), "revenue_amt":row.get("REVENUE_AMT", "").strip(), "source": "IRS EO BMF", } except Exception as exc: log.warning("_lookup_ein_csv_live url=%s error=%s", url, exc) record_failure_sync("irs_bmf") continue return None