emma_quick_search

Instructions

Implementation Reference

• server.py:1202-1248 (handler)

  The `emma_quick_search` handler function in the _dispatch method. It accepts a 'query' parameter, navigates to EMMA's QuickSearch/Results page using Playwright (headless Chromium), scrapes both issuer and issue result tables, and returns structured issuer_matches and issue_matches with URLs.

  if name == "emma_quick_search":
      query = args["query"].strip()
      if not query:
          return {"error": "query is required"}
      url = f"{EMMA_BASE}/QuickSearch/Results?quickSearchText={query.replace(' ', '%20')}"
      async with new_page() as page:
          await page.goto(url, wait_until=NAV_WAIT, timeout=NAV_TIMEOUT_MS)
          try:
              await page.wait_for_selector("#lvIssuers tbody tr, #lvIssues tbody tr, .no-data", timeout=15000)
          except Exception:
              pass
          await page.wait_for_timeout(500)
          data = await page.evaluate(
              """() => {
                  const pull = id => {
                      const t = document.getElementById(id);
                      if (!t) return {head: [], rows: []};
                      const head = Array.from(t.querySelectorAll('thead th')).map(th => th.innerText.trim());
                      const rows = Array.from(t.querySelectorAll('tbody tr')).map(tr => {
                          const cells = Array.from(tr.cells).map(td => td.innerText.trim());
                          const a = tr.querySelector('a[href*="QuickSearch/Navigate"], a[href*="IssuerHomePage/Issuer"], a[href*="IssueView/Details"], a[href*="IssuerView/IssuerDetails"], a[href*="Security/Details"]');
                          return {cells, link: a ? a.href : null};
                      });
                      return {head, rows};
                  };
                  return {issuers: pull('lvIssuers'), issues: pull('lvIssues')};
              }"""
          )
      limit = int(args.get("limit", 25))
  
      def _rowdicts(block: dict, link_key: str) -> list[dict]:
          out = []
          for r in block.get("rows", []):
              item = structured(block.get("head", []), [r.get("cells", [])])
              if not item:
                  continue
              row = item[0]
              if r.get("link"):
                  row[link_key] = r["link"]
              out.append(row)
          return out
  
      return {
          "query": query,
          "issuer_matches": _rowdicts(data["issuers"], "issuer_url")[:limit],
          "issue_matches": _rowdicts(data["issues"], "issue_url")[:limit],
      }

• server.py:590-615 (registration)

  Tool registration for 'emma_quick_search' including its name, description (starting point for obligor/issuer searches), and inputSchema requiring 'query' (natural-language search string) with optional 'limit' parameter.

  Tool(
      name="emma_quick_search",
      description=(
          "**START HERE** when the user names a specific obligor, issuer, "
          "or keyword. Uses EMMA's global quick search (the same box in "
          "EMMA's header) which indexes BOTH long-form and obligor-in-"
          "parens names — so 'Childrens Hospital' finds "
          "'CALIFORNIA HEALTH FACILITIES FINANCING AUTHORITY (LUCILE "
          "SALTER PACKARD CHILDRENS HOSPITAL AT STANFORD)' without "
          "needing the conduit-issuer chase. Returns matching issuers "
          "(across all states) and matching specific issues. Each "
          "issuer row has a URL you can feed into "
          "get_issuer_outstanding_bonds."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "Natural-language search: obligor name, issuer name, CUSIP, state, or keyword",
              },
              "limit": {"type": "integer", "default": 25},
          },
          "required": ["query"],
      },
  ),

• server.py:887-899 (handler)

  The call_tool wrapper that dispatches to _dispatch for 'emma_quick_search', applies caching based on name+arguments, and wraps errors into JSON envelopes.

  @server.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      cache_key = f"{name}:{json.dumps(arguments, sort_keys=True)}"
      if (cached := cache_get(cache_key)) is not None:
          return [TextContent(type="text", text=cached)]
      try:
          result = await _dispatch(name, arguments)
      except Exception as e:
          log(f"tool '{name}' failed: {e}\n{traceback.format_exc()}")
          return [TextContent(type="text", text=json.dumps({"error": str(e), "tool": name}))]
      text = json.dumps(result, indent=2, default=str)
      cache_put(cache_key, text)
      return [TextContent(type="text", text=text)]