instruction.md
# 식품 안전 체크
## What this skill does
식약처/식품안전나라 공개 표면을 **`k-skill-proxy` 경유**로 조회해 식품 안전 정보를 확인한다.
조회 가능한 공식 데이터:
- **건강기능식품 기능성 원료 인정현황** (I-0040) — 원료가 공식 인정됐는지, 1일 섭취량, 섭취시 주의사항
- **건강기능식품 개별인정형 정보** (I-0050) — 개별 인정 원료의 기능성, 섭취량 상·하한, 주의사항
- **건강기능식품 품목제조 신고사항** (I0030) — 신고된 제품의 원재료, 기능성, 섭취 주의사항, 기준규격 (고시형 원료 포함)
- **검사부적합 현황(국내)** (I2620) — 국내 유통 식품 부적합 판정 이력
- **부적합 식품 목록** — 공공데이터포털 경유
- **회수·판매중지 공개 목록** (I0490) — 식품안전나라 경유
하지만 사용자가 복통, 설사, 발진 같은 증상을 말하면 **바로 단정하지 말고 먼저 되묻는다.**
- 누가 먹었는지 (본인/아이/임산부/고령자)
- 무엇을 언제 얼마나 먹었는지
- 같이 먹은 음식/술/약
- 현재 증상과 시작 시점
- 기저질환, 임신, 알레르기
- red flag (`혈변`, `탈수`, `호흡곤란`, `의식저하`, `심한 복통/고열`)
red flag 가 있으면 식품 조회보다 **즉시 응급실·119·의료진 안내**가 우선이다.
## When to use
- "이 음식 먹어도 괜찮니?" → 건강기능식품 원료 인정현황 + 검사부적합 조회
- "차전자피를 먹어도 되나?" → 기능성 원료 인정현황에서 1일 섭취량·주의사항 확인
- "이 김밥 먹고 배가 아픈데 회수 이력 있나?" → 회수·부적합 목록 조회
- "식약처 공식 부적합 식품 목록에서 제품명 확인해줘"
- "식품안전나라 공개 회수 목록에서 업체명으로 찾아줘"
## Prerequisites
- 인터넷 연결
- `python3`
- `@nomadamas/k-skill` CLI에 번들된 `scripts/mfds_food_safety.py` helper
- `k-skill-proxy`의 food-safety route들이 있는 hosted/self-host 프록시에 접근 가능할 것
## Credential requirements
- 사용자 측 **필수** 시크릿 없음.
- `KSKILL_PROXY_BASE_URL` — self-host·별도 프록시를 쓸 때만 설정. 비우면 기본 hosted `https://k-skill-proxy.nomadamas.org` 를 사용한다.
- `DATA_GO_KR_API_KEY` 는 **프록시 운영 서버** 환경에서 부적합 식품 live 조회용으로만 둔다.
- `FOODSAFETYKOREA_API_KEY` 는 **프록시 운영 서버** 환경에서 식품안전나라 live 조회용으로만 둔다. 없으면 public sample feed로 fallback 할 수 있다.
- FOODSAFETYKOREA_API_KEY 발급: `https://www.foodsafetykorea.go.kr` 에서 회원가입 후 OpenAPI 이용신청. 하나의 키로 I-0040, I-0050, I0490, I2620 등 모든 서비스 호출 가능.
## Mandatory interview first
증상/섭취상황이 언급되면 결론을 말하기 전에 먼저 되묻는다.
권장 첫 질문 예시:
- `누가 무엇을 언제 얼마나 먹었는지, 지금 복통/구토/설사/발진 같은 증상이 있는지 먼저 알려주세요.`
- `호흡곤란, 혈변, 심한 탈수, 의식저하, 심한 복통/고열이 있으면 즉시 응급실이나 119가 우선입니다.`
## Official surfaces
- 공공데이터포털 문서: `https://www.data.go.kr/data/15056516/openapi.do`
- 부적합 식품 endpoint: `https://apis.data.go.kr/1471000/PrsecImproptFoodInfoService03/getPrsecImproptFoodList01`
- 식품안전나라 회수·판매중지 (I0490): `https://www.foodsafetykorea.go.kr/api/openApiInfo.do?menu_grp=MENU_GRP31&menu_no=661&show_cnt=10&start_idx=1&svc_no=I0490&svc_type_cd=API_TYPE06`
- 식품안전나라 회수 sample: `https://openapi.foodsafetykorea.go.kr/api/sample/I0490/json/1/5`
- 건강기능식품 기능성 원료 인정현황 (I-0040): `https://www.foodsafetykorea.go.kr/api/openApiInfo.do?menu_grp=MENU_GRP31&menu_no=661&show_cnt=10&start_idx=1&svc_no=I-0040&svc_type_cd=API_TYPE06`
- 건강기능식품 기능성 원료 sample: `https://openapi.foodsafetykorea.go.kr/api/sample/I-0040/json/1/5`
- 건강기능식품 개별인정형 정보 (I-0050): `https://www.foodsafetykorea.go.kr/api/openApiInfo.do?menu_grp=MENU_GRP31&menu_no=661&show_cnt=10&start_idx=1&svc_no=I-0050&svc_type_cd=API_TYPE06`
- 건강기능식품 개별인정형 sample: `https://openapi.foodsafetykorea.go.kr/api/sample/I-0050/json/1/5`
- 검사부적합(국내) (I2620): `https://www.foodsafetykorea.go.kr/api/openApiInfo.do?menu_grp=MENU_GRP31&menu_no=661&show_cnt=10&start_idx=1&svc_no=I2620&svc_type_cd=API_TYPE06`
- 검사부적합(국내) sample: `https://openapi.foodsafetykorea.go.kr/api/sample/I2620/json/1/5`
- 건강기능식품 품목제조 신고사항 (I0030): `https://www.foodsafetykorea.go.kr/api/openApiInfo.do?menu_grp=MENU_GRP31&menu_no=661&show_cnt=10&start_idx=1&svc_no=I0030&svc_type_cd=API_TYPE06`
- 건강기능식품 품목제조 신고 sample: `https://openapi.foodsafetykorea.go.kr/api/sample/I0030/json/1/5`
- 프록시 route: `GET /v1/mfds/food-safety/search`
- 프록시 route: `GET /v1/mfds/food-safety/health-food-ingredient`
- 프록시 route: `GET /v1/mfds/food-safety/product-report`
- 프록시 route: `GET /v1/mfds/food-safety/inspection-fail`
## Workflow
1. 증상/섭취상황이 있으면 인터뷰를 먼저 진행한다.
2. red flag 가 있으면 즉시 응급 안내로 전환한다.
3. "이거 먹어도 되나?" 류 질문:
- `/v1/mfds/food-safety/product-report` 로 건강기능식품 품목제조 신고사항을 조회한다 (원재료, 기능성, 섭취 주의사항, 기준규격). 고시형 원료(차전자피 등)는 여기서 확인.
- `/v1/mfds/food-safety/health-food-ingredient` 로 건강기능식품 기능성 원료 인정현황을 조회한다 (개별인정형 원료의 1일 섭취량, 주의사항).
- `/v1/mfds/food-safety/inspection-fail` 로 국내 검사부적합 이력을 확인한다.
- `/v1/mfds/food-safety/search` 로 회수·부적합 공개 목록도 함께 조회한다.
4. 제품명, 업체명, 기능성, 섭취량, 주의사항, 부적합 사유를 짧게 정리하고, 먹어도 되는지 단정하지 않는다.
5. 프록시가 `FOODSAFETYKOREA_API_KEY` 없이 동작 중이면 결과가 sample feed 기반일 수 있음을 warnings 로 확인한다.
## CLI examples
```bash
npx -y @nomadamas/k-skill@0 exec mfds-food-safety scripts/mfds_food_safety.py -- interview \
--question "이 김밥 먹어도 되나요?" \
--symptoms "복통과 설사"
```
```bash
npx -y @nomadamas/k-skill@0 exec mfds-food-safety scripts/mfds_food_safety.py -- search --query "김밥" --limit 5
```
```bash
npx -y @nomadamas/k-skill@0 exec mfds-food-safety scripts/mfds_food_safety.py -- product-report --query "차전자피" --limit 5
```
```bash
npx -y @nomadamas/k-skill@0 exec mfds-food-safety scripts/mfds_food_safety.py -- health-food-ingredient --query "스타놀" --limit 5
```
```bash
npx -y @nomadamas/k-skill@0 exec mfds-food-safety scripts/mfds_food_safety.py -- inspection-fail --query "쪽갓" --limit 5
```
## Response policy
- 이 스킬은 **직접 진단**을 하지 않는다.
- 이 스킬은 **식중독 진단**이나 **섭취 허가/금지의 최종 판정**을 하지 않는다.
- 공식 공개 목록에 있는 사실만 전달한다.
- 증상이 있는 질문은 인터뷰 없이 바로 답하지 않는다.
- red flag 또는 고위험군이면 의료진 상담을 우선 권고한다.
## Done when
- 증상 또는 섭취상황을 먼저 되물었다.
- red flag 여부를 확인했다.
- 프록시 route를 통해 공식 공개 목록에서 제품명 또는 업체명 기준 결과를 최소 1건 이상 찾았거나, 없다고 분명히 알렸다.
- 제품명, 업체명, 공개사유/부적합 사유, 공개일자를 포함한 요약을 제공했다.
scripts/mfds_food_safety.py
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from html import unescape
from typing import Any
PROXY_BASE_URL_ENV_VAR = "KSKILL_PROXY_BASE_URL"
DEFAULT_PROXY_BASE_URL = "https://k-skill-proxy.nomadamas.org"
PROXY_DOWN_MSG = "설정된 k-skill-proxy 서버가 응답하지 않습니다. 잠시 후 재시도하거나 운영자에게 문의하세요."
PROXY_KEY_NOT_CONFIGURED_MSG = "k-skill-proxy에 필요한 API 키가 설정되어 있지 않습니다. 운영자에게 문의하세요."
class ApiError(RuntimeError):
def __init__(self, message: str, *, status_code: int | None = None, url: str | None = None):
super().__init__(message)
self.status_code = status_code
self.url = url
def summarize_text(value: Any) -> str:
if value is None:
return ""
text = unescape(str(value))
text = re.sub(r"<[^>]+>", " ", text)
text = re.sub(r"\s+", " ", text).strip()
return text
def resolve_proxy_base_url(explicit_base_url: str | None = None, env: dict[str, str] | None = None) -> str:
env = env or os.environ
candidate = summarize_text(explicit_base_url or env.get(PROXY_BASE_URL_ENV_VAR))
if candidate.casefold() in {"off", "false", "0", "disable", "disabled", "none"}:
raise ValueError("KSKILL_PROXY_BASE_URL 가 비활성화되어 있습니다.")
if candidate and candidate != "replace-me":
return candidate.rstrip("/")
return DEFAULT_PROXY_BASE_URL
def build_food_interview(question: str | None = None, symptoms: str | None = None) -> dict[str, Any]:
return {
"domain": "food",
"question": summarize_text(question),
"symptoms": summarize_text(symptoms),
"must_ask": [
"누가 먹었거나 먹으려는지 알려주세요. (본인/아이/임산부/고령자)",
"무엇을 언제 먹었는지, 얼마나 먹었는지 알려주세요.",
"같이 먹은 음식이나 술, 복용 중인 약이 있는지 알려주세요.",
"복통·구토·설사·발진 같은 증상이 언제부터 시작됐는지 알려주세요.",
"기저질환, 임신 여부, 알레르기 여부를 알려주세요.",
],
"red_flags": [
"호흡곤란, 입술·혀 붓기 같은 급성 알레르기 반응",
"혈변 또는 검은변",
"심한 탈수, 소변 감소, 계속되는 구토",
"의식저하, 고열, 심한 복통",
],
"urgent_action": "red flag 가 있으면 식품 조회보다 즉시 응급실·119·의료진 연결을 우선하세요.",
"policy": "이 helper 는 공식 식품 안전정보 조회 전에 반드시 되묻기 흐름을 제공하며, 먹어도 되는지 단정하지 않습니다.",
}
def normalize_food_recall_row(row: dict[str, Any]) -> dict[str, Any]:
return {
"source": "foodsafetykorea_recall",
"product_name": summarize_text(row.get("product_name") or row.get("PRDLST_NM") or row.get("PRDTNM")),
"company_name": summarize_text(row.get("company_name") or row.get("BSSH_NM") or row.get("BSSHNM")),
"reason": summarize_text(row.get("reason") or row.get("RTRVLPRVNS")),
"created_at": summarize_text(row.get("created_at") or row.get("CRET_DTM")),
"distribution_deadline": summarize_text(row.get("distribution_deadline") or row.get("DISTBTMLMT")),
"category": summarize_text(row.get("category") or row.get("PRDLST_TYPE") or row.get("PRDLST_CD_NM")),
}
def normalize_improper_food_item(item: dict[str, Any]) -> dict[str, Any]:
reason_parts = [
summarize_text(item.get("reason") or item.get("IMPROPT_ITM")),
summarize_text(item.get("INSPCT_RESULT")),
]
return {
"source": "mfds_improper_food",
"product_name": summarize_text(item.get("product_name") or item.get("PRDUCT")),
"company_name": summarize_text(item.get("company_name") or item.get("ENTRPS")),
"reason": "; ".join(part for part in reason_parts if part),
"created_at": summarize_text(item.get("created_at") or item.get("REGIST_DT")),
"category": summarize_text(item.get("category") or item.get("FOOD_TY")),
}
def filter_food_items(items: list[dict[str, Any]], query: str) -> list[dict[str, Any]]:
needle = summarize_text(query).casefold()
if not needle:
return items
product_matches = [item for item in items if needle in summarize_text(item.get("product_name")).casefold()]
if product_matches:
return product_matches
company_matches = [item for item in items if needle in summarize_text(item.get("company_name")).casefold()]
if company_matches:
return company_matches
return [item for item in items if needle in summarize_text(item.get("reason")).casefold()]
def read_json_response(request: urllib.request.Request | str) -> dict[str, Any]:
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
try:
payload = json.loads(body)
except json.JSONDecodeError:
payload = None
if error.code == 503 and isinstance(payload, dict) and payload.get("error") == "upstream_not_configured":
raise ApiError(PROXY_KEY_NOT_CONFIGURED_MSG, status_code=error.code, url=getattr(error, "url", None)) from error
if isinstance(payload, dict) and payload.get("message"):
raise ApiError(str(payload["message"]), status_code=error.code, url=getattr(error, "url", None)) from error
raise ApiError(
f"MFDS food proxy request failed with HTTP {error.code}",
status_code=error.code,
url=getattr(error, "url", None),
) from error
except urllib.error.URLError as error:
raise ApiError(f"{PROXY_DOWN_MSG} (상세: {error.reason})") from error
def search_food_safety(
query: str,
*,
limit: int = 10,
base_url: str | None = None,
request_json: Any = read_json_response,
) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
url = f"{resolved_base_url}/v1/mfds/food-safety/search"
params = urllib.parse.urlencode({"query": query, "limit": str(limit)})
request = urllib.request.Request(
f"{url}?{params}",
headers={"Accept": "application/json", "User-Agent": "k-skill-mfds/1.0"},
)
return request_json(request)
def search_health_food_ingredient(
query: str,
*,
limit: int = 10,
base_url: str | None = None,
request_json: Any = read_json_response,
) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
url = f"{resolved_base_url}/v1/mfds/food-safety/health-food-ingredient"
params = urllib.parse.urlencode({"query": query, "limit": str(limit)})
request = urllib.request.Request(
f"{url}?{params}",
headers={"Accept": "application/json", "User-Agent": "k-skill-mfds/1.0"},
)
return request_json(request)
def search_product_report(
query: str,
*,
limit: int = 10,
base_url: str | None = None,
request_json: Any = read_json_response,
) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
url = f"{resolved_base_url}/v1/mfds/food-safety/product-report"
params = urllib.parse.urlencode({"query": query, "limit": str(limit)})
request = urllib.request.Request(
f"{url}?{params}",
headers={"Accept": "application/json", "User-Agent": "k-skill-mfds/1.0"},
)
return request_json(request)
def search_inspection_fail(
query: str,
*,
limit: int = 10,
base_url: str | None = None,
request_json: Any = read_json_response,
) -> dict[str, Any]:
resolved_base_url = resolve_proxy_base_url(base_url)
url = f"{resolved_base_url}/v1/mfds/food-safety/inspection-fail"
params = urllib.parse.urlencode({"query": query, "limit": str(limit)})
request = urllib.request.Request(
f"{url}?{params}",
headers={"Accept": "application/json", "User-Agent": "k-skill-mfds/1.0"},
)
return request_json(request)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="MFDS food-safety helper")
subparsers = parser.add_subparsers(dest="command", required=True)
interview = subparsers.add_parser("interview", help="print the mandatory symptom follow-up interview")
interview.add_argument("--question", default="")
interview.add_argument("--symptoms", default="")
search = subparsers.add_parser("search", help="search official food-safety records through k-skill-proxy")
search.add_argument("--query", required=True)
search.add_argument("--limit", type=int, default=10)
search.add_argument("--proxy-base-url")
product_report = subparsers.add_parser("product-report", help="search health food product manufacturing reports")
product_report.add_argument("--query", required=True)
product_report.add_argument("--limit", type=int, default=10)
product_report.add_argument("--proxy-base-url")
ingredient = subparsers.add_parser("health-food-ingredient", help="search health food ingredient recognition status")
ingredient.add_argument("--query", required=True)
ingredient.add_argument("--limit", type=int, default=10)
ingredient.add_argument("--proxy-base-url")
inspection = subparsers.add_parser("inspection-fail", help="search domestic inspection failure records")
inspection.add_argument("--query", required=True)
inspection.add_argument("--limit", type=int, default=10)
inspection.add_argument("--proxy-base-url")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
if args.command == "interview":
print(json.dumps(build_food_interview(question=args.question, symptoms=args.symptoms), ensure_ascii=False, indent=2))
return 0
if args.command == "search":
try:
payload = search_food_safety(args.query, limit=args.limit, base_url=args.proxy_base_url)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
if args.command == "product-report":
try:
payload = search_product_report(args.query, limit=args.limit, base_url=args.proxy_base_url)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
if args.command == "health-food-ingredient":
try:
payload = search_health_food_ingredient(args.query, limit=args.limit, base_url=args.proxy_base_url)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
if args.command == "inspection-fail":
try:
payload = search_inspection_fail(args.query, limit=args.limit, base_url=args.proxy_base_url)
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0
except (ValueError, ApiError) as error:
print(json.dumps({"error": str(error)}, ensure_ascii=False, indent=2), file=sys.stderr)
return 1
return 1
if __name__ == "__main__":
raise SystemExit(main())