Stage 0: SAP(ZAA_ICF)에서 패키지 단위로 프로그램 소스 수집 (ingest/from_sap.py)
- sap/: abap-mcp 의 catalog/sap_client/mcp_server 복사. import 와 .env 탐색만 이 저장소에 맞춤 - ingest/from_sap.py: 패키지명 → GET_PROGRAM_LIST → GET_PROGRAM_SOURCE(+Include) → data/raw/*.txt (normalize 가 읽는 수집 JSON 형식 그대로). 받은 파일은 건너뛰고 --force 로 재수집 - normalize: 응답의 TCODE_LIST 를 tcodes.jsonl 로 (from_dir 와 같은 모양, 로더가 적재) - .env.example 에 SAP_URL/SAP_USER/SAP_PASS, tests/test_from_sap.py Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
byeongwook.choi
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
11ae3629b2
commit
921b8ce229
@@ -0,0 +1,332 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ZAA_ICF (ZCL_ZAA_AGENT_API) MCP 서버.
|
||||
|
||||
catalog.py 의 16개 메서드를 MCP 툴로 노출한다 (툴 이름 = 메서드명 소문자, 파라미터 = IV_ 접두사 뗀 소문자).
|
||||
SAP 호출은 sap_client.sap_call (../abap-api-tester 의 웹 테스터와 같은 코드).
|
||||
|
||||
실행: python mcp_server.py (stdio — Claude Code / Claude Desktop 용)
|
||||
python mcp_server.py --http 8766 (streamable-http, http://127.0.0.1:8766/mcp)
|
||||
접속정보: .env 또는 환경변수 SAP_URL / SAP_USER / SAP_PASS (sap_client 가 읽는다)
|
||||
|
||||
환경변수 (선택):
|
||||
ABAP_MCP_DEFAULT_MAX_ROWS 목록 툴에서 max_rows 생략 시 상한 (기본 200). LLM 컨텍스트 보호용
|
||||
ABAP_MCP_MAX_OUTPUT_CHARS 툴 응답 JSON 최대 문자 수 (기본 120000). 넘치면 소스 → 행 순으로 잘라내고 알린다
|
||||
ABAP_MCP_TIMEOUT SAP 호출 타임아웃 초 (기본 180)
|
||||
"""
|
||||
import inspect
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Annotated, Any
|
||||
|
||||
import anyio
|
||||
from pydantic import Field
|
||||
from mcp.server.mcpserver import MCPServer
|
||||
from mcp.server.mcpserver.exceptions import ToolError
|
||||
|
||||
from catalog import CATALOG, CODE_FIELDS, FIELDS, LIST_FIELDS, METHODS, VALUE_MAPS
|
||||
import sap_client
|
||||
from sap_client import CONFIG, sap_call
|
||||
|
||||
DEFAULT_MAX_ROWS = int(os.environ.get("ABAP_MCP_DEFAULT_MAX_ROWS", "200"))
|
||||
MAX_OUTPUT_CHARS = int(os.environ.get("ABAP_MCP_MAX_OUTPUT_CHARS", "120000"))
|
||||
TIMEOUT = int(os.environ.get("ABAP_MCP_TIMEOUT", "180"))
|
||||
|
||||
# 소스를 돌려주는 메서드에는 줄 범위 파라미터를 붙여 LLM 이 페이지 단위로 읽게 한다
|
||||
SOURCE_METHODS = {"GET_PROGRAM_SOURCE", "GET_VERSION_SOURCE", "GET_FUNCTION_DETAIL"}
|
||||
|
||||
INSTRUCTIONS = """SAP ABAP 시스템(ZCL_ZAA_AGENT_API, 읽기 전용)을 탐색하는 툴 모음. 모든 툴은 조회만 하며 아무것도 바꾸지 않는다.
|
||||
|
||||
탐색 흐름
|
||||
- 어디에 뭐가 있나: get_package_list → get_object_list_by_package(OBJECT 값이 다음 툴을 정한다: PROG→get_program_source, FUGR→get_function_list, TABL→get_table_fields, DTEL/DOMA→get_type_info, TRAN→get_tcode_info)
|
||||
- 사용자가 'VA01 화면' 처럼 T코드로 말하면 get_tcode_info 로 프로그램을 찾은 뒤 get_program_source(with_include=true)
|
||||
- 업무 용어('회사코드')만 알 때: get_field_list_by_text → 데이터엘리먼트/테이블 → get_table_fields / get_type_info
|
||||
- 영향도 분석: get_where_used_list. 변경 이력: get_cts_list → get_cts_object_list, get_version_list → get_version_source 두 버전 diff
|
||||
|
||||
응답 읽는 법
|
||||
- 모든 응답은 {"return": {"type","message","total_rows"}, ...결과}. type S=성공, W=경고(대개 결과가 상한에서 잘림), E=입력 오류/미존재, X=통신·파싱 오류.
|
||||
- 목록 툴은 max_rows 를 생략하면 %d 건까지만 가져온다. 잘리면 type=W 와 total_rows 로 전체 건수를 알려주니, 더 필요하면 max_rows 를 올리거나 0(전건, 안전 상한 50,000)으로 부른다. 페이지(offset) 개념은 없다. 정렬이 유형/이름순이라 잘리면 뒤쪽 유형이 통째로 빠질 수 있다.
|
||||
- 응답이 너무 크면 이 서버가 소스 → 목록 행 순으로 잘라내고 "_truncated" 에 알린다. 소스는 line_from/line_to 로 나눠 읽는다.
|
||||
- 이름류 입력은 서버가 대문자로 바꾸지만 검색어(search_text)는 원문 그대로 부분 일치·대소문자 구분. 날짜는 YYYY-MM-DD.
|
||||
- 코드값 뜻이 궁금하면 explain_fields 로 필드 사전을 본다 (예: SUBC, OBJECT, TRFUNCTION).
|
||||
""" % DEFAULT_MAX_ROWS
|
||||
|
||||
mcp = MCPServer(
|
||||
"abap-sap",
|
||||
title="SAP ABAP 탐색 (ZAA_ICF)",
|
||||
instructions=INSTRUCTIONS,
|
||||
version="1.0.0",
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 카탈로그 → 툴 시그니처
|
||||
def _arg_name(p: dict) -> str:
|
||||
n = p["name"]
|
||||
return (n[3:] if n.startswith("IV_") else n).lower()
|
||||
|
||||
|
||||
def _py_type(p: dict):
|
||||
return {"number": int, "flag": bool}.get(p["type"], str)
|
||||
|
||||
|
||||
def _param_desc(p: dict) -> str:
|
||||
d = p.get("label", "")
|
||||
if p.get("desc"):
|
||||
d += ". " + p["desc"]
|
||||
if p.get("placeholder") and p["type"] != "flag":
|
||||
d += f" (예: {p['placeholder']})"
|
||||
if p["type"] == "date":
|
||||
d += ". 형식 YYYY-MM-DD"
|
||||
if p["name"] == "IV_MAX_ROWS":
|
||||
d = f"최대 건수. 생략 시 {DEFAULT_MAX_ROWS}, 0 이면 전건(안전 상한 50,000). 잘리면 return.type=W + total_rows"
|
||||
return d
|
||||
|
||||
|
||||
def _tool_description(m: dict) -> str:
|
||||
lines = [f"[{m['title']}] {m['summary']}", f"언제: {m['when']}"]
|
||||
for n in m.get("notes", []):
|
||||
lines.append(f"주의: {n}")
|
||||
ex = [s for s in m.get("samples", []) if s["params"] and "→ E" not in s["label"]][:3]
|
||||
if ex:
|
||||
lines.append("예: " + " / ".join(
|
||||
f"{s['label']}: {json.dumps({_arg_name({'name': k}): v for k, v in s['params'].items()}, ensure_ascii=False)}"
|
||||
for s in ex))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _build_signature(m: dict) -> inspect.Signature:
|
||||
params = []
|
||||
for p in m["params"]:
|
||||
name, typ = _arg_name(p), _py_type(p)
|
||||
if p.get("required"):
|
||||
ann, default = Annotated[typ, Field(description=_param_desc(p))], inspect.Parameter.empty
|
||||
elif typ is bool:
|
||||
ann, default = Annotated[bool, Field(description=_param_desc(p))], False
|
||||
else:
|
||||
ann, default = Annotated[typ | None, Field(description=_param_desc(p))], None
|
||||
params.append(inspect.Parameter(name, inspect.Parameter.KEYWORD_ONLY, default=default, annotation=ann))
|
||||
if m["name"] in SOURCE_METHODS:
|
||||
params.append(inspect.Parameter("line_from", inspect.Parameter.KEYWORD_ONLY, default=None,
|
||||
annotation=Annotated[int | None, Field(description="소스 시작 줄(1부터). 긴 소스를 나눠 읽을 때. 생략 시 처음부터")]))
|
||||
params.append(inspect.Parameter("line_to", inspect.Parameter.KEYWORD_ONLY, default=None,
|
||||
annotation=Annotated[int | None, Field(description="소스 끝 줄(포함). 생략 시 끝까지")]))
|
||||
# 필수 파라미터가 앞에 오도록 (KEYWORD_ONLY 라 순서 제약은 없지만 스키마 가독성)
|
||||
params.sort(key=lambda x: x.default is not inspect.Parameter.empty)
|
||||
return inspect.Signature(params, return_annotation=dict[str, Any])
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 호출 + 응답 다듬기
|
||||
def _to_sap_params(m: dict, kwargs: dict) -> dict:
|
||||
out = {}
|
||||
for p in m["params"]:
|
||||
v = kwargs.get(_arg_name(p))
|
||||
if p["type"] == "flag":
|
||||
if v:
|
||||
out[p["name"]] = "X"
|
||||
elif p["name"] == "IV_MAX_ROWS":
|
||||
if v is None:
|
||||
out[p["name"]] = DEFAULT_MAX_ROWS
|
||||
elif v > 0:
|
||||
out[p["name"]] = v
|
||||
# 0 → 전건 (파라미터 생략)
|
||||
elif v is not None and str(v) != "":
|
||||
out[p["name"]] = v
|
||||
return out
|
||||
|
||||
|
||||
def _slice_sources(obj, line_from, line_to):
|
||||
"""CODE_FIELDS 를 줄 범위로 자르고 *_LINES(전체 줄 수) 를 붙인다. include/screen 목록 안의 소스도 같이."""
|
||||
if isinstance(obj, dict):
|
||||
for k in list(obj.keys()):
|
||||
v = obj[k]
|
||||
if k in CODE_FIELDS and isinstance(v, str):
|
||||
lines = v.split("\n")
|
||||
obj[k + "_LINES"] = len(lines)
|
||||
lo = max(1, line_from or 1)
|
||||
hi = min(len(lines), line_to or len(lines))
|
||||
if lo > 1 or hi < len(lines):
|
||||
obj[k] = "\n".join(lines[lo - 1:hi])
|
||||
obj[k + "_RANGE"] = f"{lo}-{hi}"
|
||||
else:
|
||||
_slice_sources(v, line_from, line_to)
|
||||
elif isinstance(obj, list):
|
||||
for x in obj:
|
||||
_slice_sources(x, line_from, line_to)
|
||||
|
||||
|
||||
def _size(obj) -> int:
|
||||
return len(json.dumps(obj, ensure_ascii=False))
|
||||
|
||||
|
||||
def _shrink(data: dict) -> list[str]:
|
||||
"""응답이 MAX_OUTPUT_CHARS 를 넘으면 그때그때 가장 큰 항목(소스 필드 또는 목록)을 절반으로 줄인다.
|
||||
|
||||
잘린 자리에는 *_CUT / *_TOTAL 표식을 남기고, 마지막에 표식을 모아 설명 문장 목록으로 돌려준다.
|
||||
소스는 1,000자 아래로, 목록은 1행 아래로는 줄이지 않는다.
|
||||
"""
|
||||
if _size(data) <= MAX_OUTPUT_CHARS:
|
||||
return []
|
||||
|
||||
def candidates(obj, path=""):
|
||||
if isinstance(obj, dict):
|
||||
for k, v in obj.items():
|
||||
if k in CODE_FIELDS and isinstance(v, str) and len(v) >= 2000:
|
||||
yield len(v), obj, k, path + "/" + k
|
||||
elif k in LIST_FIELDS and isinstance(v, list) and len(v) > 1:
|
||||
yield _size(v), obj, k, path + "/" + k
|
||||
yield from candidates(v, path + "/" + k)
|
||||
else:
|
||||
yield from candidates(v, path + "/" + k)
|
||||
elif isinstance(obj, list):
|
||||
for i, x in enumerate(obj):
|
||||
yield from candidates(x, f"{path}[{i}]")
|
||||
|
||||
for _ in range(400):
|
||||
if _size(data) <= MAX_OUTPUT_CHARS:
|
||||
break
|
||||
cands = sorted(candidates(data), key=lambda t: -t[0])
|
||||
if not cands:
|
||||
break
|
||||
_, holder, key, _ = cands[0]
|
||||
v = holder[key]
|
||||
if isinstance(v, str):
|
||||
keep = v[: max(1000, len(v) // 2)]
|
||||
keep = keep[: keep.rfind("\n") + 1] or keep
|
||||
holder[key + "_CUT"] = holder.get(key + "_CUT", 0) + (v.count("\n") - keep.count("\n"))
|
||||
holder[key] = keep
|
||||
else:
|
||||
holder.setdefault(key + "_TOTAL", len(v))
|
||||
holder[key] = v[: max(1, len(v) // 2)]
|
||||
|
||||
# 표식을 모아 설명으로
|
||||
notes, cut_sources = [], 0
|
||||
|
||||
def collect(obj, path=""):
|
||||
nonlocal cut_sources
|
||||
if isinstance(obj, dict):
|
||||
for k, v in obj.items():
|
||||
if k.endswith("_TOTAL") and k[:-6] in LIST_FIELDS:
|
||||
notes.append(f"{(path + '/' + k[:-6]).lstrip('/')}: {v}행 중 앞 {len(obj[k[:-6]])}행만 남김 (max_rows 를 줄이거나 조건을 좁힐 것)")
|
||||
elif k.endswith("_CUT") and k[:-4] in CODE_FIELDS:
|
||||
if path.count("[") == 0:
|
||||
notes.append(f"{(path + '/' + k[:-4]).lstrip('/')}: 뒤쪽 {v}줄 생략 (line_from/line_to 로 이어서 읽을 것)")
|
||||
else:
|
||||
cut_sources += 1
|
||||
else:
|
||||
collect(v, path + "/" + k)
|
||||
elif isinstance(obj, list):
|
||||
for i, x in enumerate(obj):
|
||||
collect(x, f"{path}[{i}]")
|
||||
|
||||
collect(data)
|
||||
if cut_sources:
|
||||
notes.append(f"목록 안 소스 {cut_sources}개의 뒷부분 생략 (해당 Include/화면을 개별 툴로 다시 조회하거나 line_from/line_to 사용)")
|
||||
notes.insert(0, f"응답이 {MAX_OUTPUT_CHARS:,}자를 넘어 이 서버가 잘라냈습니다 (ABAP_MCP_MAX_OUTPUT_CHARS)")
|
||||
return notes
|
||||
|
||||
|
||||
def _call(m: dict, kwargs: dict) -> dict:
|
||||
if not CONFIG["user"] or not CONFIG["password"]:
|
||||
raise ToolError("SAP 접속 정보가 없습니다. .env 또는 환경변수 SAP_USER / SAP_PASS 를 설정하세요.")
|
||||
params = _to_sap_params(m, kwargs)
|
||||
res = sap_call(m["name"], params, timeout=TIMEOUT)
|
||||
|
||||
if res["error"] and res["parsed"] is None: # 네트워크/인증/파싱 — 결과 자체가 없음
|
||||
hint = ""
|
||||
if res.get("http_status") == 401:
|
||||
hint = " (인증 실패: SAP_USER / SAP_PASS 확인)"
|
||||
raise ToolError(f"{m['name']} 호출 실패{hint}: {res['error']} [HTTP {res.get('http_status')}]")
|
||||
|
||||
parsed = res["parsed"] if isinstance(res["parsed"], dict) else {"RESULT": res["parsed"]}
|
||||
ret_raw = parsed.pop("RETURN", {}) if isinstance(parsed.get("RETURN"), dict) else {}
|
||||
out: dict[str, Any] = {
|
||||
"return": {
|
||||
"type": res["return"]["type"],
|
||||
"message": res["return"]["message"],
|
||||
},
|
||||
}
|
||||
try:
|
||||
out["return"]["total_rows"] = int(ret_raw.get("TOTAL_ROWS") or 0)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
if res["error"]: # ICF 에러 봉투 (<error>) 등: 본문은 있으나 오류
|
||||
out["return"]["error"] = res["error"]
|
||||
out["return"]["http_status"] = res.get("http_status")
|
||||
out.update(parsed)
|
||||
|
||||
if m["name"] in SOURCE_METHODS:
|
||||
_slice_sources(out, kwargs.get("line_from"), kwargs.get("line_to"))
|
||||
notes = _shrink(out)
|
||||
if notes:
|
||||
out["_truncated"] = notes
|
||||
out["_meta"] = {"sap_method": m["name"], "sap_params": params, "ms": res["ms"], "raw_bytes": res.get("raw_size")}
|
||||
return out
|
||||
|
||||
|
||||
def _make_tool(m: dict):
|
||||
async def tool(**kwargs):
|
||||
return await anyio.to_thread.run_sync(_call, m, kwargs)
|
||||
|
||||
tool.__name__ = m["name"].lower()
|
||||
tool.__doc__ = _tool_description(m)
|
||||
tool.__signature__ = _build_signature(m) # inspect.signature 가 이것을 읽는다
|
||||
tool.__annotations__ = {p.name: p.annotation for p in tool.__signature__.parameters.values()}
|
||||
tool.__annotations__["return"] = dict[str, Any]
|
||||
return tool
|
||||
|
||||
|
||||
for _m in METHODS:
|
||||
mcp.add_tool(_make_tool(_m), name=_m["name"].lower(), title=_m["title"])
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 보조 툴
|
||||
@mcp.tool(title="필드 사전 / 코드값 해석")
|
||||
def explain_fields(
|
||||
fields: Annotated[list[str] | None, Field(description="응답 필드명 목록 (예: [\"SUBC\", \"OBJECT\", \"TRFUNCTION\"]). 생략하면 전체 사전")] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""응답 컬럼의 한국어 뜻과 코드값 해석표를 돌려준다 (SAP 을 모르는 사용자에게 결과를 설명할 때).
|
||||
|
||||
예: SUBC=1 은 실행 프로그램, OBJECT=FUGR 은 함수그룹, TRFUNCTION=K 는 워크벤치 요청.
|
||||
"""
|
||||
names = [f.upper() for f in fields] if fields else sorted(FIELDS)
|
||||
out = {}
|
||||
for n in names:
|
||||
entry: dict[str, Any] = {}
|
||||
if n in FIELDS:
|
||||
entry["label"], entry["desc"] = FIELDS[n]
|
||||
if n in VALUE_MAPS:
|
||||
entry["values"] = VALUE_MAPS[n]
|
||||
out[n] = entry or "사전에 없음"
|
||||
return out
|
||||
|
||||
|
||||
@mcp.tool(title="접속 상태 / 호출 규약")
|
||||
def sap_connection_info() -> dict[str, Any]:
|
||||
"""현재 SAP 접속 대상과 호출 규약을 보여준다 (비밀번호는 제외). 툴이 X 오류를 낼 때 먼저 확인."""
|
||||
return {
|
||||
"base_url": CONFIG["base_url"],
|
||||
"user": CONFIG["user"] or "(미설정)",
|
||||
"password_set": bool(CONFIG["password"]),
|
||||
"verify_ssl": CONFIG["verify_ssl"],
|
||||
"default_max_rows": DEFAULT_MAX_ROWS,
|
||||
"max_output_chars": MAX_OUTPUT_CHARS,
|
||||
"conventions": CATALOG["conventions"],
|
||||
"tools": [m["name"].lower() for m in METHODS],
|
||||
}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 실행
|
||||
def main():
|
||||
args = sys.argv[1:]
|
||||
if args and args[0] == "--http":
|
||||
port = int(args[1]) if len(args) > 1 else 8766
|
||||
print(f"abap-sap MCP (streamable-http) -> http://127.0.0.1:{port}/mcp", file=sys.stderr)
|
||||
mcp.run(transport="streamable-http", host="127.0.0.1", port=port)
|
||||
else:
|
||||
# stdio: stdout 은 프로토콜 채널이므로 로그는 stderr 로만
|
||||
print(f"abap-sap MCP (stdio) SAP={CONFIG['base_url']} user={CONFIG['user'] or '(미설정)'}", file=sys.stderr)
|
||||
mcp.run(transport="stdio")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user