fda_510k_recent

Instructions

Implementation Reference

• trends_mcp.py:1134-1140 (schema)

  Pydantic model FDA510kInput: validates query, days (1-365), max_results (1-100), and response_format for the fda_510k_recent tool.

  class FDA510kInput(BaseModel):
      model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
      query: str | None = Field(None, max_length=300)
      days: int = Field(30, ge=1, le=365)
      max_results: int = Field(20, ge=1, le=100)
      response_format: ResponseFormat = ResponseFormat.MARKDOWN

• trends_mcp.py:1188-1201 (registration)

  Tool registration via @_maybe_tool decorator for source='fda_510k', name='fda_510k_recent'. The _maybe_tool function gates registration based on whether the source is in ENABLED_SOURCES.

  @_maybe_tool(
      source="fda_510k",
      name="fda_510k_recent",
      description=(
          "Recent FDA 510(k) clearances via openFDA. Date filter is always applied. "
          "openFDA uses token-exact matching on string fields — for partial name "
          "matches use wildcards (e.g. `device_name:mammo*` not `device_name:mammography`)."
      ),
      annotations={
          "readOnlyHint": True,
          "destructiveHint": False,
          "openWorldHint": True,
          "idempotentHint": True,
      },

• trends_mcp.py:1203-1243 (handler)

  Main handler function for fda_510k_recent. Uses openFDA API to search 510(k) clearances filtered by decision_date range and optional query. Builds a custom URL with +AND+ search syntax, fetches JSON with TTL_STATIC cache, renders markdown via _render_510k_md, and handles 404 (no results) gracefully.

  async def fda_510k_recent(
      query: str | None = None,
      days: int = 30,
      max_results: int = 20,
      response_format: ResponseFormat = ResponseFormat.MARKDOWN,
  ) -> str:
      try:
          args = FDA510kInput(
              query=query, days=days, max_results=max_results, response_format=response_format
          )
          end = _utc_now().strftime("%Y%m%d")
          start = (_utc_now() - timedelta(days=args.days)).strftime("%Y%m%d")
          date_clause = f"decision_date:[{start}+TO+{end}]"
          parts = [date_clause]
          if args.query:
              parts.append(args.query)
          search = _build_openfda_search(parts)
          params: dict[str, Any] = {
              "search": search,
              "limit": args.max_results,
              "sort": "decision_date:desc",
          }
          api_key = os.environ.get("OPENFDA_API_KEY")
          if api_key:
              params["api_key"] = api_key
          # openFDA's `+AND+` must NOT be percent-encoded; pass as a manually-built URL.
          url = f"{OPENFDA_510K}?search={search}&limit={params['limit']}&sort=decision_date:desc"
          if api_key:
              url += f"&api_key={api_key}"
          data = await _http_get_json(url, ttl=TTL_STATIC)
          items = data.get("results", []) if isinstance(data, dict) else []
          header = f"FDA 510(k) — 최근 {args.days}일 ({len(items)}건)"
          return _format(items, args.response_format, render_md=lambda x: _render_510k_md(x, header))
      except httpx.HTTPStatusError as e:
          # openFDA returns 404 when zero results — show empty list instead of error.
          if e.response.status_code == 404:
              header = f"FDA 510(k) — 최근 {args.days}일 (0건)"
              return _format([], args.response_format, render_md=lambda x: _render_510k_md(x, header))
          return _handle_error(e, "fda_510k_recent")
      except Exception as e:
          return _handle_error(e, "fda_510k_recent")

• trends_mcp.py:1152-1166 (helper)

  Markdown renderer _render_510k_md: formats 510(k) results into a structured markdown table with device name (linked to FDA's pmn page), applicant, decision description, date, and product code.

  def _render_510k_md(items: list[dict[str, Any]], header: str) -> str:
      if not items:
          return f"# {header}\n\n_결과 없음_"
      lines = [f"# {header}", f"_총 {len(items)}건_", ""]
      for i, r in enumerate(items, 1):
          knum = r.get("k_number", "")
          url = f"https://www.accessdata.fda.gov/scripts/cdrh/cfdocs/cfpmn/pmn.cfm?ID={knum}" if knum else ""
          link = f"[{knum}]({url})" if url else knum
          lines.append(
              f"## {i}. {r.get('device_name', '?')} — {link}\n"
              f"- 신청자: {r.get('applicant', '?')}\n"
              f"- 결정: {r.get('decision_description') or r.get('decision_code', '?')} · "
              f"{r.get('decision_date', '?')} · 제품코드 `{r.get('product_code', '?')}`\n"
          )
      return "\n".join(lines)

• trends_mcp.py:1184-1185 (helper)

  Helper _build_openfda_search: joins search parts with openFDA's +AND+ syntax for building the query string.

  def _build_openfda_search(parts: list[str]) -> str:
      return "+AND+".join(parts) if parts else ""