joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refs #732: add dynamic HMM HTML report MCP tool

Browse Source
This commit is contained in:
devmrko
2026-08-10 15:23:36 +09:00
parent cc2c3e3e25
commit c0c5a36ba0
12 changed files with 747 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

357
ai-web-agent-console/app.py
View File

@@ -2633,6 +2633,14 @@ def _mcp_response_payload(mcp_result: Any) -> Mapping[str, Any]:
if isinstance(mcp_result, Mapping):
response = mcp_result.get("response")
if isinstance(response, Mapping):
nested_result = response.get("result")
if isinstance(nested_result, str):
try:
parsed = json.loads(nested_result)
except ValueError:
parsed = None
if isinstance(parsed, Mapping):
return parsed
return response
return mcp_result
return {}
@@ -2643,6 +2651,19 @@ def _mcp_generated_sql(mcp_result: Any) -> str:
return str(payload.get("generatedSql") or payload.get("generated_sql") or "").strip()
def _mcp_rendered_html(mcp_result: Any) -> str:
"""Extract an HTML artifact without coupling the UI to a tool name."""
payload = _mcp_response_payload(mcp_result)
candidate = payload.get("html") or payload.get("rendered_html")
if not isinstance(candidate, str):
return ""
html_text = candidate.strip()
if len(html_text) > 1_000_000 or "<" not in html_text or ">" not in html_text:
return ""
return html_text
def _mcp_items(mcp_result: Any) -> list[Any]:
payload = _mcp_response_payload(mcp_result)
items = payload.get("items")
@@ -2692,11 +2713,91 @@ def _agent_tool_catalog(
if isinstance(properties, Mapping)
else [],
"read_only": route.tool.read_only,
"consumes_previous_result": _is_presentation_route(route),
}
)
return catalog, routes
def _is_presentation_route(route: RoutedMcpTool) -> bool:
"""Identify a renderer from its discovered description and input schema."""
properties = route.tool.schema.get("properties")
names = (
[str(name).casefold() for name in properties]
if isinstance(properties, Mapping)
else []
)
has_payload_input = any(
any(marker in name for marker in ("report", "payload", "render_data"))
and any(marker in name for marker in ("json", "data", "payload"))
for name in names
)
description = f"{route.tool.name} {route.tool.description}".casefold()
has_presentation_intent = any(
marker in description
for marker in ("render", "html", "dashboard", "presentation", "report")
)
has_prior_result_contract = any(
marker in description
for marker in (
"supplied",
"already-authorized",
"previous result",
"input data",
"후속 처리",
"조회 결과",
"선행 결과",
)
)
return has_payload_input and has_presentation_intent and has_prior_result_contract
def _question_requests_presentation(question: str) -> bool:
text = str(question or "").casefold()
return any(
marker in text
for marker in (
"리포트",
"보고서",
"대시보드",
"차트",
"그래프",
"html",
"화면으로 만들어",
"문서로 만들어",
)
)
def _presentation_route(
routes_by_key: Mapping[str, RoutedMcpTool],
attempted_route_keys: set[str],
) -> tuple[str, RoutedMcpTool] | None:
return next(
(
(key, route)
for key, route in routes_by_key.items()
if key not in attempted_route_keys and _is_presentation_route(route)
),
None,
)
def _primary_data_route(
routes_by_key: Mapping[str, RoutedMcpTool],
attempted_route_keys: set[str],
) -> tuple[str, RoutedMcpTool] | None:
return next(
(
(key, route)
for key, route in routes_by_key.items()
if key not in attempted_route_keys and not _is_presentation_route(route)
),
None,
)
def _complex_reasoning_model_profile(default_model_profile: str) -> str:
configured = (
os.environ.get(COMPLEX_REASONING_MODEL_PROFILE_ENV)
@@ -2740,6 +2841,120 @@ def _clean_agent_tool_query(value: object, fallback: str) -> str:
return " ".join(cleaned).strip() or fallback
def _camel_case_key(value: object) -> str:
text = str(value or "")
if "_" not in text:
return text[:1].lower() + text[1:]
parts = [part for part in text.casefold().split("_") if part]
return (
parts[0] + "".join(part[:1].upper() + part[1:] for part in parts[1:])
if parts
else text
)
def _normalize_presentation_value(value: Any) -> Any:
if isinstance(value, Mapping):
return {
_camel_case_key(key): _normalize_presentation_value(item)
for key, item in value.items()
}
if isinstance(value, list):
return [_normalize_presentation_value(item) for item in value]
return value
def _presentation_payload(question: str, steps: list[Mapping[str, Any]]) -> dict[str, Any]:
"""Build a renderer payload only from the preceding authorized tool result."""
source = next(
(
step
for step in reversed(steps)
if not bool(step.get("consumes_previous_result"))
),
{},
)
raw_result = source.get("mcp_result", {}) if isinstance(source, Mapping) else {}
payload = _mcp_response_payload(raw_result)
source_rows = _mcp_items(raw_result)
if not source_rows:
for key in ("results", "rows"):
candidate = payload.get(key) if isinstance(payload, Mapping) else None
if isinstance(candidate, list):
source_rows = candidate
break
rows = [
_normalize_presentation_value(row)
for row in source_rows
if isinstance(row, Mapping)
]
requester = re.search(r"\bE\d{4,}\b", str(question or ""), flags=re.IGNORECASE)
source_tool = str(source.get("tool_name") or "MCP data query")
return {
"report": {
"id": "MCP-REPORT",
"category": "MCP Business Intelligence",
"title": str(question or "MCP 결과 리포트"),
"question": str(question or ""),
"requestedBy": requester.group(0).upper() if requester else "-",
"generatedAt": datetime.now(timezone.utc).isoformat(),
"answer": (
f"{source_tool}에서 권한 범위 내 결과 {len(rows)}건을 받아 "
"리포트로 구성했습니다."
),
"execution": {
"server": str(source.get("server_id") or ""),
"tool": source_tool,
"calls": len(steps) + 1,
},
"evidence": [f"MCP {source_tool} 결과 {len(rows)}건 반환"],
"limitation": "표시 값은 선행 MCP 데이터 조회 결과만 사용했습니다.",
},
"rows": rows,
}
def _build_mcp_arguments(
*,
route: RoutedMcpTool,
tool_query: str,
limit: int,
preferred_tool: str,
steps: list[Mapping[str, Any]],
) -> dict[str, Any]:
if not _is_presentation_route(route):
return build_mcp_tool_arguments(
route.tool,
tool_query,
int(limit),
preferred_tool=preferred_tool,
)
properties = route.tool.schema.get("properties")
if not isinstance(properties, Mapping):
raise McpToolRouterError("렌더링 MCP tool의 입력 스키마가 없습니다.")
input_name = next(
(
str(name)
for name in properties
if "report" in str(name).casefold()
and any(
marker in str(name).casefold()
for marker in ("json", "payload", "data")
)
),
"",
)
if not input_name:
raise McpToolRouterError("렌더링 MCP tool이 report payload 입력을 선언하지 않았습니다.")
return {
input_name: json.dumps(
_presentation_payload(tool_query, steps),
ensure_ascii=False,
)
}
def _mcp_has_actionable_result(mcp_result: Any) -> bool:
if _mcp_generated_sql(mcp_result) or _mcp_items(mcp_result):
return True
@@ -3231,6 +3446,29 @@ def plan_mcp_execution_mode(
"reason": f"사용자 선택: {override}",
"model_profile": "",
}
presentation_routes = [route for route in routed_tools if _is_presentation_route(route)]
if _question_requests_presentation(question) and presentation_routes:
data_routes = [route for route in routed_tools if not _is_presentation_route(route)]
primary = data_routes[0] if data_routes else fallback
if len(data_routes) > 1:
try:
primary = route_mcp_tool_across_servers_with_llm(
data_routes,
question,
router_model_profile=model_profile_key,
)
except McpToolRouterError:
pass
if primary is not fallback or len(routed_tools) > 1:
return {
"mode": "agent",
"route_key": _route_key(primary.server_id, primary.tool.name),
"reason": (
"리포트·차트·HTML 요청이며, 발견된 렌더링 도구가 이전 구조화 "
"결과를 입력으로 선언했습니다. 데이터 조회 후 렌더링을 순차 실행합니다."
),
"model_profile": model_profile_key,
}
if len(routed_tools) <= 1:
return {
"mode": "single",
@@ -3276,7 +3514,10 @@ def plan_mcp_execution_mode(
"the user's Korean question should be answered with one MCP "
"tool call or with a multi-tool ReAct loop. Prefer mode=single "
"unless the question explicitly requires comparing, combining, "
"or validating evidence across different MCP tools/sources. "
"or validating evidence across different MCP tools/sources. A route "
"whose consumes_previous_result flag is true is a presentation renderer: "
"when the user asks for a report, dashboard, chart, document, or HTML, "
"choose mode=agent so a data route runs first and that renderer runs next. "
"Return only JSON matching the schema. Never request or expose "
"bearer tokens."
),
@@ -3290,7 +3531,8 @@ def plan_mcp_execution_mode(
),
"agent_policy": (
"use agent only for cross-source comparison, contract "
"plus terms/document search, or multi-step validation"
"plus terms/document search, multi-step validation, or a "
"data-result-to-presentation rendering sequence"
),
},
ensure_ascii=False,
@@ -3360,6 +3602,10 @@ def _plan_agent_step(
"If no tool has been called yet, choose action=call_tool. "
"After each observation, decide whether another tool call is needed "
"or action=final_answer is enough. Do not expose or request bearer tokens. "
"A route with consumes_previous_result=true is a renderer. For a report, "
"dashboard, chart, document, or HTML request, call a data route first, "
"then call that renderer using the previous structured result; do not "
"finish with text only before the renderer has been attempted. "
"tool_query must be plain natural-language query text only; do not include "
"argument labels such as limit:, prompt:, query:, top_k:, or candidate_k:. "
"Do not write SQL. Preserve identifiers exactly. If the user says "
@@ -3417,6 +3663,7 @@ def run_mcp_agent_loop(
bearer_token: str,
limit: int,
model_profile_key: str,
initial_route_key: str = "",
progress_callback: Any = None,
) -> dict[str, Any]:
tool_catalog, routes_by_key = _agent_tool_catalog(routed_tools)
@@ -3431,8 +3678,21 @@ def run_mcp_agent_loop(
for step_no in range(1, MAX_AGENT_TOOL_STEPS + 1):
forced_plan = None
wants_presentation = _question_requests_presentation(question)
if wants_presentation and _presentation_route(routes_by_key, attempted_route_keys):
forced_plan = (
(
(initial_route_key, routes_by_key[initial_route_key])
if not observations
and initial_route_key in routes_by_key
and not _is_presentation_route(routes_by_key[initial_route_key])
else _primary_data_route(routes_by_key, attempted_route_keys)
)
if not observations
else _presentation_route(routes_by_key, attempted_route_keys)
)
if step_no == 1 and _question_needs_cross_source(question):
forced_plan = _select_unvisited_route(
forced_plan = forced_plan or _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
@@ -3522,36 +3782,48 @@ def run_mcp_agent_loop(
tool_query = _clean_agent_tool_query(plan.get("tool_query"), question)
if action == "final_answer" and observations:
forced = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
renderer = (
_presentation_route(routes_by_key, attempted_route_keys)
if wants_presentation
else None
)
if forced is None or not _question_needs_cross_source(question):
stop_reason = "planner가 추가 MCP 호출이 불필요하다고 판단했습니다."
break
route_key, route = forced
action = "call_tool"
tool_query = _fallback_tool_query_for_route(
question,
route,
observations,
)
thought = (
f"{thought} / 비교·약관 질문인데 미호출 MCP route가 있어 "
f"{route_key} 호출로 전환합니다."
).strip(" /")
if renderer is not None:
route_key, route = renderer
action = "call_tool"
tool_query = question
thought = "사용자 요청에 따른 이전 구조화 결과의 리포트 렌더링"
else:
forced = _select_unvisited_route(
question,
routes_by_key,
attempted_route_keys,
)
if forced is None or not _question_needs_cross_source(question):
stop_reason = "planner가 추가 MCP 호출이 불필요하다고 판단했습니다."
break
route_key, route = forced
action = "call_tool"
tool_query = _fallback_tool_query_for_route(
question,
route,
observations,
)
thought = (
f"{thought} / 비교·약관 질문인데 미호출 MCP route가 있어 "
f"{route_key} 호출로 전환합니다."
).strip(" /")
if action != "call_tool" and not observations:
action = "call_tool"
route = routes_by_key.get(route_key)
if route is None:
route = next(iter(routes_by_key.values()))
route_key = _route_key(route.server_id, route.tool.name)
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
if not _is_presentation_route(route):
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
is_distinct_vector_query = (
_is_vector_route(route)
and tool_query not in completed_vector_queries
@@ -3576,11 +3848,12 @@ def run_mcp_agent_loop(
route,
observations,
)
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
if not _is_presentation_route(route):
tool_query = append_query_contract_guidance(
tool_query,
original_question=question,
tool_name=route.tool.name,
)
thought = (
f"{thought} / 중복 MCP route {repeated_route_key} 대신 "
f"미호출 route {route_key}로 전환합니다."
@@ -3590,11 +3863,12 @@ def run_mcp_agent_loop(
raise PublicMcpError("선택된 MCP 서버 설정을 찾지 못했습니다.")
started = perf_counter()
arguments = build_mcp_tool_arguments(
route.tool,
tool_query,
int(limit),
arguments = _build_mcp_arguments(
route=route,
tool_query=question if _is_presentation_route(route) else tool_query,
limit=int(limit),
preferred_tool=server.default_tool,
steps=steps,
)
raw_result = call_tool(
base_url=server.endpoint_url,
@@ -3618,6 +3892,7 @@ def run_mcp_agent_loop(
"mcp_result": mcp_result,
"result_summary": summary,
"elapsed_seconds": round(elapsed, 3),
"consumes_previous_result": _is_presentation_route(route),
}
steps.append(step)
observations.append(
@@ -3642,6 +3917,9 @@ def run_mcp_agent_loop(
actionable_route_keys.add(route_key)
if progress_callback:
progress_callback(step)
if _is_presentation_route(route):
stop_reason = "이전 구조화 결과를 렌더링 MCP tool로 전달했습니다."
break
if last is None:
raise PublicMcpError("MCP agent가 실행한 tool 호출이 없습니다.")
@@ -3666,6 +3944,11 @@ def _render_mcp_result_sections(
execution_events = details.get("execution_events")
generated_sql = _mcp_generated_sql(mcp_result)
items = _mcp_items(mcp_result)
rendered_html = _mcp_rendered_html(mcp_result)
if rendered_html:
st.markdown("#### 생성된 리포트")
components.html(rendered_html, height=1420, scrolling=True)
if isinstance(execution_events, list) and execution_events:
with st.expander(f"처리 시간 로그 · {len(execution_events)}건"):
@@ -5031,6 +5314,10 @@ def _process_submitted_question(
bearer_token=bearer_token,
limit=int(limit),
model_profile_key=reasoning_model_profile,
initial_route_key=_route_key(
selected_route.server_id,
selected_route.tool.name,
),
progress_callback=on_agent_step,
)
selected_server = agent_result["server"]