instruction.md
# EV Charger Nearby
## What this skill does
환경부 전기차 충전소 API의 충전소 정보와 충전기 상태를 조회한다. 조회 전용이며 예약, 결제, 회원 인증, 충전 시작/중지는 자동화하지 않는다.
기본 호출은 hosted `k-skill-proxy`를 사용하므로 사용자는 API 키가 필요 없다. `--direct`를 명시한 경우에만 사용자의 `KSKILL_EV_CHARGER_API_KEY`, 이후 `DATA_GO_KR_API_KEY`를 찾는다.
## Official access path
1. 실시간 충전소 정보: 공공데이터포털 데이터셋 `15076352`, `getChargerInfo`
2. 실시간 충전기 상태: 같은 데이터셋의 `getChargerStatus`
3. 정적/수동 fallback: 전국전기차충전소표준데이터 데이터셋 `15013115`의 포털 제공 파일
`15013115`는 실시간 상태 API가 아니다. 포털 다운로드 화면에서 사용자가 직접 받은 표준 CSV를 정적 참고자료로만 사용하며, 문서화되지 않은 CSV URL을 추측하지 않는다.
## Inputs and commands
설치된 skill 디렉터리에서 실행한다.
```bash
npx -y @nomadamas/k-skill@0 exec ev-charger-nearby scripts/ev_charger.py -- info --location '서울 강남구' --num-of-rows 10
npx -y @nomadamas/k-skill@0 exec ev-charger-nearby scripts/ev_charger.py -- status --stat-id ME000001 --num-of-rows 10
npx -y @nomadamas/k-skill@0 exec ev-charger-nearby scripts/ev_charger.py -- status --zcode 11 --limit-yn Y --period 10 --json
```
공통 필터는 `zcode`, `zscode`, `statId`, `chgerId`다. hosted proxy의 `info`는 고유하게 식별되는 `location`을 행정구역 코드로 변환하며, 모호하거나 찾을 수 없는 위치와 충돌하는 명시적 코드는 거부한다. `status`는 `limitYn`과 `period`를 추가로 허용한다. 페이지 기본값은 `pageNo=1`, `numOfRows=10`이며 `numOfRows` 범위는 10~9999다.
직접 호출:
```bash
npx -y @nomadamas/k-skill@0 exec ev-charger-nearby scripts/ev_charger.py -- info --zcode 11 --zscode 11680 --direct
npx -y @nomadamas/k-skill@0 exec ev-charger-nearby scripts/ev_charger.py -- status --stat-id ME000001 --direct --dry-run
```
직접 호출은 위치 텍스트를 행정구역 코드로 변환하지 않으므로 `--location` 대신 `--zcode`/`--zscode`를 사용한다. `--dry-run`은 URL을 보여주되 키를 `REDACTED`로 가린다.
## Credential rules
- hosted proxy: 사용자 키 불필요
- direct 환경변수 우선순위: `KSKILL_EV_CHARGER_API_KEY` -> `DATA_GO_KR_API_KEY`
- dotenv fallback: `~/.config/k-skill/secrets.env`에서 같은 순서
- 키를 로그, 답변, dry-run URL, 저장소 파일에 노출하지 않는다
공공데이터포털 인증키를 이미 갖고 있어도 데이터셋 `15076352` 상세 페이지에서 **별도 활용신청**이 필요하다. 이 API는 자동승인 대상이지만 승인/활성화가 완료된 뒤 호출해야 한다.
## Workflow and fallback order
1. 위치/지역 요청은 `info`로 충전소 주소, 운영기관, 충전기 유형과 식별자를 찾는다.
2. 찾은 `statId`/`chgerId`로 `status`를 호출해 현재 상태와 갱신시각을 확인한다.
3. live info가 실패하면 동일 조건을 좁혀 한 번 재시도한다.
4. live info는 성공했지만 status가 실패하면 정보 결과를 제공하면서 상태 확인 실패를 분리해 알린다.
5. live API 전체가 불가능할 때만 데이터셋 `15013115`의 사용자가 직접 내려받은 표준 CSV를 정적/수동 fallback으로 안내한다.
## Output
기본은 한국어 요약, `--json`은 프록시/상류 JSON을 출력한다. 가능한 경우 다음 필드를 포함한다.
- 충전소명, 주소, 위치 설명
- 운영기관
- `statId`, `chgerId`
- 충전기 유형과 현재 상태
- 상태 갱신시각
상태 코드는 원문 값을 보존한다. 상태가 없거나 오래된 경우 추정하지 않는다.
## Failure modes
- `upstream_not_configured`: hosted/self-host 프록시에 `DATA_GO_KR_API_KEY`가 없음
- key/auth rejection: 데이터셋 `15076352` 활용신청 승인 여부 확인
- proxy down: 프록시 장애 메시지를 보여주고 잠시 후 재시도
- empty result: 조건을 좁히거나 `zcode`/`zscode`/`statId`를 확인
- invalid/non-JSON/XML response: upstream 응답 오류로 분리하고 결과로 캐시하지 않음
- static CSV fallback: 실시간 상태로 표현하지 않고 파일 기준일을 함께 명시
## Done when
- 충전소 정보와 가능한 현재 상태를 각각 확인했다.
- 정보 조회와 상태 조회의 시점/실패를 구분했다.
- 키, 예약, 결제, 충전 제어를 노출하거나 자동화하지 않았다.
scripts/ev_charger.py
#!/usr/bin/env python3
"""Read-only EV charger info/status helper using stdlib only."""
from __future__ import annotations
import argparse
import json
import os
import pathlib
import sys
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, Optional
DEFAULT_PROXY_BASE_URL = "https://k-skill-proxy.nomadamas.org"
DEFAULT_SECRETS_PATH = pathlib.Path("~/.config/k-skill/secrets.env").expanduser()
UPSTREAM_BASE_URL = "https://apis.data.go.kr/B552584/EvCharger"
OPERATIONS = {"info": "getChargerInfo", "status": "getChargerStatus"}
PROXY_DOWN_MSG = "설정된 k-skill-proxy 프록시 서버가 응답하지 않습니다. 잠시 후 재시도하거나 운영자에게 문의하세요."
PROXY_NOT_CONFIGURED_MSG = "k-skill-proxy에 EV 충전소 API 키가 설정되어 있지 않습니다. 운영자에게 문의하세요."
class HelperError(RuntimeError):
pass
def load_secrets(path: pathlib.Path) -> Dict[str, str]:
values: Dict[str, str] = {}
try:
lines = path.read_text(encoding="utf-8").splitlines()
except OSError:
return values
for raw in lines:
line = raw.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
value = value.strip().strip('"').strip("'")
values[key.strip()] = value
return values
def resolve_api_key(args: argparse.Namespace) -> Optional[str]:
env_key = os.environ.get("KSKILL_EV_CHARGER_API_KEY") or os.environ.get("DATA_GO_KR_API_KEY")
if env_key and env_key.strip():
return env_key.strip()
secrets = load_secrets(pathlib.Path(args.secrets_path).expanduser())
value = secrets.get("KSKILL_EV_CHARGER_API_KEY") or secrets.get("DATA_GO_KR_API_KEY")
return value.strip() if value and value.strip() else None
def _bounded_integer(value: int, label: str, minimum: int, maximum: int) -> int:
if value < minimum or value > maximum:
raise HelperError(f"{label} 값은 {minimum}~{maximum} 범위여야 합니다.")
return value
def _validate_code(value: Optional[str], label: str, maximum: int, digits: Optional[int] = None) -> Optional[str]:
if value is None:
return None
text = value.strip()
if not text:
return None
if len(text) > maximum or (digits is not None and (len(text) != digits or not text.isdigit())):
raise HelperError(f"올바른 {label} 값을 입력하세요.")
if digits is None and not all(ch.isalnum() or ch in "_-" for ch in text):
raise HelperError(f"올바른 {label} 값을 입력하세요.")
return text
def build_query(args: argparse.Namespace) -> Dict[str, Any]:
query: Dict[str, Any] = {
"pageNo": _bounded_integer(args.page_no, "pageNo", 1, 100000),
"numOfRows": _bounded_integer(args.num_of_rows, "numOfRows", 10, 9999),
}
for key, maximum, digits in (("zcode", 2, 2), ("zscode", 5, 5), ("stat_id", 40, None), ("chger_id", 10, None)):
value = _validate_code(getattr(args, key), key.replace("_", ""), maximum, digits)
if value is not None:
query[{"stat_id": "statId", "chger_id": "chgerId"}.get(key, key)] = value
if args.command == "info" and args.location:
location = args.location.strip()
if len(location) > 100:
raise HelperError("location 값은 100자 이하여야 합니다.")
if args.direct:
raise HelperError("--direct에서는 location을 지원하지 않습니다. --zcode와 --zscode를 사용하세요.")
query["location"] = location
if args.command == "status":
if args.limit_yn:
limit_yn = args.limit_yn.strip().upper()
if limit_yn not in {"Y", "N"}:
raise HelperError("limitYn 값은 Y 또는 N이어야 합니다.")
query["limitYn"] = limit_yn
if args.period is not None:
query["period"] = _bounded_integer(args.period, "period", 1, 10)
return query
def build_url(args: argparse.Namespace, query: Dict[str, Any], api_key: Optional[str]) -> str:
if args.direct:
if not api_key:
raise HelperError(
"KSKILL_EV_CHARGER_API_KEY 또는 DATA_GO_KR_API_KEY가 없습니다. "
"공공데이터포털 데이터셋 15076352 활용신청 후 키를 환경변수나 ~/.config/k-skill/secrets.env에 설정하세요."
)
direct_query = dict(query)
direct_query["serviceKey"] = api_key
direct_query["dataType"] = "JSON"
return f"{UPSTREAM_BASE_URL}/{OPERATIONS[args.command]}?{urllib.parse.urlencode(direct_query)}"
return f"{args.proxy_base_url.rstrip('/')}/v1/ev-charger/{args.command}?{urllib.parse.urlencode(query)}"
def http_get_json(url: str, timeout: int, via_proxy: bool) -> Dict[str, Any]:
request = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": "k-skill/ev-charger-nearby"})
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
raw = response.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as error:
raw = error.read().decode("utf-8", errors="replace") if error.fp else ""
try:
payload = json.loads(raw)
except json.JSONDecodeError:
payload = {}
if via_proxy and error.code == 503 and payload.get("error") == "upstream_not_configured":
raise HelperError(PROXY_NOT_CONFIGURED_MSG) from error
message = payload.get("message") or f"API HTTP 오류: {error.code} {error.reason}"
raise HelperError(str(message)) from error
except urllib.error.URLError as error:
if via_proxy:
raise HelperError(f"{PROXY_DOWN_MSG} (상세: {error.reason})") from error
raise HelperError(f"상류 API 네트워크 오류: {error.reason}") from error
except TimeoutError as error:
target = "프록시 서버" if via_proxy else "상류 API"
raise HelperError(f"{target} 요청 시간이 초과되었습니다.") from error
try:
payload = json.loads(raw)
except json.JSONDecodeError as error:
raise HelperError("API 응답이 올바른 JSON이 아닙니다.") from error
if not isinstance(payload, dict):
raise HelperError("API 응답 형식이 올바르지 않습니다.")
return payload
def format_text(payload: Dict[str, Any]) -> str:
response = payload.get("response")
if isinstance(response, dict) and isinstance(response.get("body"), dict):
payload = response["body"]
items_value = payload.get("items") or []
if isinstance(items_value, dict):
items_value = items_value.get("item") or []
items = items_value if isinstance(items_value, list) else [items_value]
total_count = payload.get("total_count", payload.get("totalCount", len(items)))
lines = [f"전기차 충전기 조회 결과: {total_count}건"]
for item in items:
name = item.get("statNm") or item.get("statId") or "이름 없음"
address = item.get("addr") or item.get("location") or "주소 없음"
charger = item.get("chgerId") or "-"
status = item.get("stat") or "상태 미상"
lines.append(f"- {name} / {address} / 충전기 {charger} / 상태 {status}")
if not items:
lines.append("조건에 맞는 충전기 정보가 없습니다.")
return "\n".join(lines)
def add_common_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--zcode")
parser.add_argument("--zscode")
parser.add_argument("--stat-id")
parser.add_argument("--chger-id")
parser.add_argument("--page-no", type=int, default=1)
parser.add_argument("--num-of-rows", type=int, default=10)
parser.add_argument("--proxy-base-url", default=os.environ.get("KSKILL_PROXY_BASE_URL", DEFAULT_PROXY_BASE_URL))
parser.add_argument("--direct", action="store_true")
parser.add_argument("--secrets-path", default=str(DEFAULT_SECRETS_PATH))
parser.add_argument("--timeout", type=int, default=90)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--json", action="store_true")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="전기차 충전소 정보와 상태 조회")
subparsers = parser.add_subparsers(dest="command", required=True)
info = subparsers.add_parser("info", help="충전소 위치와 기본 정보 조회")
add_common_arguments(info)
info.add_argument("--location")
status = subparsers.add_parser("status", help="충전기 운영 상태 조회")
add_common_arguments(status)
status.add_argument("--limit-yn")
status.add_argument("--period", type=int)
return parser
def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace:
return build_parser().parse_args(argv)
def run(argv: Optional[list[str]] = None) -> int:
try:
args = parse_args(argv)
query = build_query(args)
api_key = resolve_api_key(args) if args.direct else None
if args.dry_run and args.direct:
url = build_url(args, query, "REDACTED")
else:
url = build_url(args, query, api_key)
if args.dry_run:
print(json.dumps({"operation": args.command, "url": url, "query": query}, ensure_ascii=False, indent=2))
return 0
payload = http_get_json(url, args.timeout, via_proxy=not args.direct)
if args.json:
print(json.dumps(payload, ensure_ascii=False, indent=2))
else:
print(format_text(payload))
return 0
except HelperError as error:
print(str(error), file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(run())
skill.json
{
"name": "ev-charger-nearby",
"description": "Use when the user asks for Korean EV charging-station locations, charger types, or current charger status by region, station, or location text. 돌쇠에서는 공식 표면을 통한 후속 액션까지 진행한다.",
"profiles": [
"proxy",
"lookup"
],
"frontmatter": "name: ev-charger-nearby\ndescription: Use when the user asks for Korean EV charging-station locations, charger types, or current charger status by region, station, or location text. 돌쇠에서는 공식 표면을 통한 후속 액션까지 진행한다.\nlicense: MIT\nmetadata:\n category: transit\n locale: ko-KR\n phase: v1"
}
SKILL.md
---
name: ev-charger-nearby
description: Use when the user asks for Korean EV charging-station locations, charger types, or current charger status by region, station, or location text. 돌쇠에서는 공식 표면을 통한 후속 액션까지 진행한다.
license: MIT
metadata:
category: transit
locale: ko-KR
phase: v1
---
# ev-charger-nearby
<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->
## Get the full instructions (required first step)
Run this and follow its output as the primary instructions for this skill:
```bash
npx -y @nomadamas/k-skill@0 instruct ev-charger-nearby
```
The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:
```bash
npx -y @nomadamas/k-skill@0 files ev-charger-nearby
```
Keep the CLI and every coding-agent skill install current (including Vercel Agent Skills) with:
```bash
npx -y @nomadamas/k-skill@0 update
```
If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/ev-charger-nearby/instruction.md.
## Hard rules even without the CLI
- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.
tests/test_ev_charger.py
import contextlib
import importlib.util
import io
import json
import os
import pathlib
import unittest
import urllib.error
from unittest import mock
ROOT = pathlib.Path(__file__).resolve().parents[2]
MODULE_PATH = ROOT / "ev-charger-nearby" / "scripts" / "ev_charger.py"
SPEC = importlib.util.spec_from_file_location("ev_charger", MODULE_PATH)
ev_charger = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(ev_charger)
class EvChargerHelperTests(unittest.TestCase):
def test_proxy_url_has_no_key_and_forces_only_server_side_json(self):
args = ev_charger.parse_args(["info", "--location", "서울", "--proxy-base-url", "https://example.test"])
query = ev_charger.build_query(args)
url = ev_charger.build_url(args, query, api_key=None)
self.assertTrue(url.startswith("https://example.test/v1/ev-charger/info?"))
self.assertNotIn("serviceKey", url)
self.assertNotIn("dataType", url)
def test_direct_url_uses_preferred_key_and_forces_json(self):
args = ev_charger.parse_args(["status", "--stat-id", "ME000001", "--direct"])
query = ev_charger.build_query(args)
url = ev_charger.build_url(args, query, api_key="secret +/==")
parsed = __import__("urllib.parse").parse.urlparse(url)
params = __import__("urllib.parse").parse.parse_qs(parsed.query)
self.assertEqual(params["serviceKey"], ["secret +/=="])
self.assertEqual(params["dataType"], ["JSON"])
self.assertTrue(parsed.path.endswith("/getChargerStatus"))
def test_direct_location_is_rejected_and_proxy_location_is_preserved(self):
proxy_args = ev_charger.parse_args(["info", "--location", "서울 강남구"])
self.assertEqual(ev_charger.build_query(proxy_args)["location"], "서울 강남구")
direct_args = ev_charger.parse_args(["info", "--location", "서울 강남구", "--direct"])
with self.assertRaisesRegex(ev_charger.HelperError, "zcode.*zscode"):
ev_charger.build_query(direct_args)
def test_pagination_and_period_follow_upstream_bounds(self):
for value in (10, 9999):
args = ev_charger.parse_args(["info", "--num-of-rows", str(value)])
self.assertEqual(ev_charger.build_query(args)["numOfRows"], value)
for value in (3, 10000):
args = ev_charger.parse_args(["info", "--num-of-rows", str(value)])
with self.assertRaisesRegex(ev_charger.HelperError, "numOfRows"):
ev_charger.build_query(args)
for value in (1, 10):
args = ev_charger.parse_args(["status", "--period", str(value)])
self.assertEqual(ev_charger.build_query(args)["period"], value)
args = ev_charger.parse_args(["status", "--period", "11"])
with self.assertRaisesRegex(ev_charger.HelperError, "period"):
ev_charger.build_query(args)
def test_direct_missing_key_reports_dataset_specific_action(self):
stderr = io.StringIO()
with mock.patch.dict(os.environ, {}, clear=True), contextlib.redirect_stderr(stderr):
code = ev_charger.run(["info", "--direct", "--secrets-path", "/tmp/missing-ev-secrets"])
self.assertEqual(code, 1)
self.assertIn("15076352", stderr.getvalue())
self.assertIn("KSKILL_EV_CHARGER_API_KEY", stderr.getvalue())
def test_key_resolution_prefers_skill_key_then_data_go_key(self):
args = ev_charger.parse_args(["info", "--direct", "--secrets-path", "/tmp/missing-ev-secrets"])
with mock.patch.dict(os.environ, {
"KSKILL_EV_CHARGER_API_KEY": "skill-key",
"DATA_GO_KR_API_KEY": "shared-key",
}, clear=True):
self.assertEqual(ev_charger.resolve_api_key(args), "skill-key")
with mock.patch.dict(os.environ, {"DATA_GO_KR_API_KEY": "shared-key"}, clear=True):
self.assertEqual(ev_charger.resolve_api_key(args), "shared-key")
def test_dry_run_redacts_direct_key(self):
stdout = io.StringIO()
with mock.patch.dict(os.environ, {"KSKILL_EV_CHARGER_API_KEY": "super-secret"}, clear=True), contextlib.redirect_stdout(stdout):
code = ev_charger.run(["status", "--stat-id", "ME000001", "--direct", "--dry-run"])
self.assertEqual(code, 0)
output = stdout.getvalue()
self.assertNotIn("super-secret", output)
self.assertIn("REDACTED", output)
def test_proxy_down_and_upstream_not_configured_have_explicit_korean_messages(self):
with mock.patch.object(ev_charger.urllib.request, "urlopen", side_effect=urllib.error.URLError("down")):
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
code = ev_charger.run(["info", "--location", "서울"])
self.assertEqual(code, 1)
self.assertIn("프록시 서버가 응답하지 않습니다", stderr.getvalue())
with mock.patch.object(ev_charger.urllib.request, "urlopen", side_effect=TimeoutError("timed out")):
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
code = ev_charger.run(["info", "--location", "서울"])
self.assertEqual(code, 1)
self.assertIn("시간이 초과", stderr.getvalue())
self.assertNotIn("Traceback", stderr.getvalue())
body = io.BytesIO(json.dumps({"error": "upstream_not_configured"}).encode("utf-8"))
error = urllib.error.HTTPError("https://example.test", 503, "Service Unavailable", {}, body)
with mock.patch.object(ev_charger.urllib.request, "urlopen", side_effect=error):
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
code = ev_charger.run(["status", "--stat-id", "ME000001"])
self.assertEqual(code, 1)
self.assertIn("API 키가 설정되어 있지 않습니다", stderr.getvalue())
def test_default_timeout_covers_slow_upstream(self):
args = ev_charger.parse_args(["info", "--location", "서울 강남구"])
self.assertGreaterEqual(args.timeout, 90)
def test_text_and_json_outputs_are_supported(self):
payload = {
"operation": "info",
"total_count": 1,
"items": [{"statNm": "시청 충전소", "addr": "서울 중구", "statId": "ME000001", "chgerId": "01", "stat": "2"}],
}
with mock.patch.object(ev_charger, "http_get_json", return_value=payload):
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
self.assertEqual(ev_charger.run(["info", "--location", "서울"]), 0)
self.assertIn("시청 충전소", stdout.getvalue())
self.assertIn("서울 중구", stdout.getvalue())
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
self.assertEqual(ev_charger.run(["info", "--location", "서울", "--json"]), 0)
self.assertEqual(json.loads(stdout.getvalue())["total_count"], 1)
def test_text_output_accepts_direct_upstream_item_envelope(self):
payload = {
"pageNo": 1,
"numOfRows": 10,
"totalCount": 1,
"items": {"item": {"statNm": "직접 호출 충전소", "addr": "서울", "chgerId": "01", "stat": "2"}},
}
self.assertIn("직접 호출 충전소", ev_charger.format_text(payload))
nested = {"response": {"body": payload}}
text = ev_charger.format_text(nested)
self.assertIn("직접 호출 충전소", text)
self.assertIn("1건", text)
if __name__ == "__main__":
unittest.main()