get_new_issue_calendar
Filter and retrieve upcoming municipal bond issues from MSRB EMMA by multiple criteria. Each issue is enriched with inferred sector classification for quick analysis.
Instructions
Upcoming municipal bond issues from MSRB EMMA — the primary competitive-intel tool for any muni banker. Each row is enriched with an inferred sector classification. Multi-axis filters: state, sector, tax_status, bank_qualified, sale_method, description_contains, min_principal, max_principal. Use this to answer questions like 'what hospital deals over $100M are pricing this week?'
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| state | No | 2-letter state code | |
| sector | No | Sector name, matched case-insensitive substring. One of: Healthcare / Hospital, Housing, Education / Schools, Water / Sewer / Utility, Transportation, Industrial Development / IDB, Tobacco Settlement, Pension / OPEB, Refunding, General Obligation, Revenue | |
| tax_status | No | 'Tax Exempt', 'Taxable', 'AMT', or 'Subject to AMT' | |
| bank_qualified | No | ||
| sale_method | No | all | |
| description_contains | No | Free-text filter on the issue description. Query is auto-expanded to EMMA abbreviations (hospital→HOSP, authority→AUTH, senior→SR, revenue→REV, etc.) — search in natural English, not telegraphic shorthand. | |
| min_principal | No | ||
| max_principal | No | ||
| limit | No | ||
| offset | No |
Implementation Reference
- server.py:369-421 (registration)Tool registration for 'get_new_issue_calendar' — defines the tool name, description, and inputSchema in the list_tools() handler.
@server.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="get_new_issue_calendar", description=( "Upcoming municipal bond issues from MSRB EMMA — the primary " "competitive-intel tool for any muni banker. Each row is enriched " "with an inferred sector classification. Multi-axis filters: state, " "sector, tax_status, bank_qualified, sale_method, description_contains, " "min_principal, max_principal. Use this to answer questions like " "'what hospital deals over $100M are pricing this week?'" ), inputSchema={ "type": "object", "properties": { "state": {"type": "string", "description": "2-letter state code"}, "sector": { "type": "string", "description": ( "Sector name, matched case-insensitive substring. One of: " "Healthcare / Hospital, Housing, Education / Schools, " "Water / Sewer / Utility, Transportation, " "Industrial Development / IDB, Tobacco Settlement, " "Pension / OPEB, Refunding, General Obligation, Revenue" ), }, "tax_status": { "type": "string", "description": "'Tax Exempt', 'Taxable', 'AMT', or 'Subject to AMT'", }, "bank_qualified": {"type": "boolean"}, "sale_method": { "type": "string", "enum": ["competitive", "negotiated", "all"], "default": "all", }, "description_contains": { "type": "string", "description": ( "Free-text filter on the issue description. Query is " "auto-expanded to EMMA abbreviations (hospital→HOSP, " "authority→AUTH, senior→SR, revenue→REV, etc.) — " "search in natural English, not telegraphic shorthand." ), }, "min_principal": {"type": "number"}, "max_principal": {"type": "number"}, "limit": {"type": "integer", "default": 25}, "offset": {"type": "integer", "default": 0}, }, }, ), - server.py:907-922 (handler)Handler dispatch for 'get_new_issue_calendar' — fetches calendar data via fetch_calendar(), filters by sale_method, applies _filter_calendar(), and returns paginated results.
if name == "get_new_issue_calendar": cal = await fetch_calendar() sale_method = (args.get("sale_method") or "all").lower() source = cal.get(sale_method, cal["all"]) rows = _filter_calendar(source, args) limit = int(args.get("limit", 25)) offset = int(args.get("offset", 0)) return { "total_matching": len(rows), "returning": min(limit, max(0, len(rows) - offset)), "filters_applied": { k: v for k, v in args.items() if v not in (None, "", "all") and k not in ("limit", "offset") }, "issues": rows[offset : offset + limit], } - server.py:762-788 (helper)fetch_calendar() — fetches the full new-issue calendar from EMMA, enriches each row with inferred sector classification, and caches the result.
async def fetch_calendar() -> dict[str, list[dict[str, str]]]: """Fetch the full new-issue calendar and return structured rows from each subtable (all / competitive / negotiated / recent sales). Enriches every row with inferred sector.""" cache_key = "_calendar" if (cached := cache_get(cache_key)) is not None: return cached tables = await load_tables(f"{EMMA_BASE}/ToolsAndResources/NewIssueCalendar") def _pull(table_id: str) -> list[dict[str, str]]: t = tables.get(table_id, {}) rows = structured(t.get("header", []), t.get("rows", [])) for r in rows: desc = r.get("Issuer Name/Issue Description", "") r["_sector"] = classify_sector(desc) r["_principal_num"] = str(_parse_amount(r.get("Principal Amount ($)", ""))) return rows payload = { "all": _pull("tblAllUpcomingIssues"), "competitive": _pull("tblCompetitiveUpcomingIssues"), "negotiated": _pull("tblNegotiatedUpcomingIssues"), "recent_sales": _pull("tblRecentSales"), } cache_put(cache_key, payload) return payload - server.py:791-823 (helper)_filter_calendar() — filters calendar rows by state, sector, tax_status, bank_qualified, description_contains, min_principal, and max_principal.
def _filter_calendar(rows: list[dict], args: dict) -> list[dict]: state = (args.get("state") or "").upper() sector = (args.get("sector") or "").lower() tax_status = (args.get("tax_status") or "").lower() bq = args.get("bank_qualified") desc = (args.get("description_contains") or "").lower() min_p = float(args.get("min_principal") or 0) max_p = float(args.get("max_principal") or 0) desc_patterns = _query_patterns(desc) if desc else [] out = [] for r in rows: if state and r.get("State", "").upper() != state: continue if sector and sector not in r.get("_sector", "").lower(): continue if tax_status and tax_status not in r.get("Tax Status", "").lower(): continue if bq is not None: row_bq = r.get("Bank Qualified", "").lower() == "yes" if bool(bq) != row_bq: continue if desc_patterns and not _match_query( r.get("Issuer Name/Issue Description", ""), desc_patterns ): continue pnum = float(r.get("_principal_num", "0") or 0) if min_p and pnum < min_p: continue if max_p and pnum > max_p: continue out.append(r) return out - server.py:382-420 (schema)inputSchema for 'get_new_issue_calendar' — defines properties: state, sector, tax_status, bank_qualified, sale_method, description_contains, min_principal, max_principal, limit, offset.
inputSchema={ "type": "object", "properties": { "state": {"type": "string", "description": "2-letter state code"}, "sector": { "type": "string", "description": ( "Sector name, matched case-insensitive substring. One of: " "Healthcare / Hospital, Housing, Education / Schools, " "Water / Sewer / Utility, Transportation, " "Industrial Development / IDB, Tobacco Settlement, " "Pension / OPEB, Refunding, General Obligation, Revenue" ), }, "tax_status": { "type": "string", "description": "'Tax Exempt', 'Taxable', 'AMT', or 'Subject to AMT'", }, "bank_qualified": {"type": "boolean"}, "sale_method": { "type": "string", "enum": ["competitive", "negotiated", "all"], "default": "all", }, "description_contains": { "type": "string", "description": ( "Free-text filter on the issue description. Query is " "auto-expanded to EMMA abbreviations (hospital→HOSP, " "authority→AUTH, senior→SR, revenue→REV, etc.) — " "search in natural English, not telegraphic shorthand." ), }, "min_principal": {"type": "number"}, "max_principal": {"type": "number"}, "limit": {"type": "integer", "default": 25}, "offset": {"type": "integer", "default": 0}, }, },