byeongwook.choiandClaude Fable 5.1
921b8ce229
- 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>
77 lines
3.2 KiB
Python
77 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""MCP 서버 연쇄 확인: stdio 로 mcp_server.py 를 띄워 툴 목록과 몇 가지 실제 호출을 점검한다.
|
|
|
|
python tools/test_mcp.py # 기본 케이스
|
|
python tools/test_mcp.py --shrink # 응답 상한을 15,000자로 낮춰 잘라내기 동작까지
|
|
|
|
접속 정보는 .env (SAP_USER / SAP_PASS). mcp 패키지 필요 (pip install -r requirements.txt).
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from mcp.client import Client
|
|
from mcp.client.stdio import StdioServerParameters
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
CASES = [
|
|
("sap_connection_info", {}),
|
|
("explain_fields", {"fields": ["SUBC", "TRFUNCTION"]}),
|
|
("get_package_list", {"max_rows": 3}), # W + total_rows
|
|
("get_tcode_info", {"tcode": "ZSCAL"}),
|
|
("get_program_source", {"program": "ZFI1000", "line_from": 1, "line_to": 5}),
|
|
("get_program_source", {"program": "SAPMZSFT0", "with_include": True, "with_screen": True, "line_to": 4}),
|
|
("get_function_detail", {"function": "ZFI_CHECK_STCD2", "line_to": 3}),
|
|
("get_version_source", {"objname": "LZLEASE01TOP", "objtype": "REPS", "versno": "00001", "line_to": 2}),
|
|
("get_where_used_list", {"obj_type": "TABL", "obj_name": "ZFIT0000"}),
|
|
("get_table_fields", {"table": "ZZZ_NOPE"}), # E 응답 (툴 오류 아님)
|
|
("get_cts_list", {"date_from": "2026-01-01"}), # 필수 누락 → 툴 오류
|
|
]
|
|
SHRINK_CASES = [
|
|
("get_program_source", {"program": "SAPMV45A", "with_include": True}), # 원문 4MB
|
|
("get_object_list_by_package", {"package": "ZFI01", "obj_type": "PROG", "max_rows": 0}),
|
|
]
|
|
|
|
|
|
def brief(d):
|
|
if isinstance(d, dict):
|
|
return {k: (f"<str {len(v)}>" if isinstance(v, str) and len(v) > 60 else brief(v)) for k, v in d.items()}
|
|
if isinstance(d, list):
|
|
return f"<list {len(d)}>" + (" first=" + json.dumps(brief(d[0]), ensure_ascii=False)[:160] if d else "")
|
|
return d
|
|
|
|
|
|
async def main(shrink: bool):
|
|
env = dict(os.environ)
|
|
if shrink:
|
|
env["ABAP_MCP_MAX_OUTPUT_CHARS"] = "15000"
|
|
params = StdioServerParameters(command=sys.executable, args=[str(ROOT / "mcp_server.py")], cwd=str(ROOT), env=env)
|
|
fails = 0
|
|
async with Client(params) as c:
|
|
tools = (await c.list_tools()).tools
|
|
print(f"tools: {len(tools)} {[t.name for t in tools]}")
|
|
for name, args in (SHRINK_CASES if shrink else CASES):
|
|
r = await c.call_tool(name, args)
|
|
txt = r.content[0].text if r.content else ""
|
|
print(f"\n=== {name} {json.dumps(args, ensure_ascii=False)} error={r.is_error} chars={len(txt):,}")
|
|
if r.is_error:
|
|
print(" ", txt.splitlines()[0][:200])
|
|
continue
|
|
d = json.loads(txt)
|
|
print(" ", json.dumps(brief(d), ensure_ascii=False)[:700])
|
|
if d.get("_truncated"):
|
|
for n in d["_truncated"]:
|
|
print(" -", n)
|
|
if shrink and len(txt) > 15000:
|
|
fails += 1
|
|
print(" !! 상한 초과")
|
|
print("\nOK" if not fails else f"\nFAIL {fails}")
|
|
return fails
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main("--shrink" in sys.argv)))
|