zhuchaowe/qwen_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update mcp-ui

Browse Source
This commit is contained in:
朱潮 2026-05-18 16:23:32 +08:00
parent fd43415a59
commit cb4a1df0b4
9 changed files with 543 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
routes/chat.py
View File

@ -114,8 +114,8 @@ async def enhanced_generate_stream_response(
chunk_args = tool_call_chunk.get("args") if isinstance(tool_call_chunk, dict) else getattr(tool_call_chunk, "args", None)
if chunk_name:
current_tool_name = chunk_name
# Always output ask_user tool calls even when tool_response is disabled
if config.tool_response or current_tool_name == 'ask_user':
# Always output ask_user and render_ui tool calls even when tool_response is disabled
if config.tool_response or current_tool_name in ('ask_user', 'render_ui'):
if chunk_name:
new_content = f"[{message_tag}] {chunk_name}\n"
if chunk_args:
@ -144,7 +144,7 @@ async def enhanced_generate_stream_response(
elif isinstance(msg, ToolMessage) and msg.content:
message_tag = "TOOL_RESPONSE"
waiting_for_answer_first_char = False
# Always output UIResource and ask_user responses even when tool_response is disabled
# Always output UIResource, ask_user and render_ui responses even when tool_response is disabled
is_ui_resource = (
msg.text
and msg.text.lstrip().startswith('{"')
@ -152,7 +152,8 @@ async def enhanced_generate_stream_response(
and ('"text/html' in msg.text or '"text/uri-list' in msg.text)
)
is_ask_user = msg.name == 'ask_user'
if config.tool_response or is_ui_resource or is_ask_user:
is_render_ui = msg.name == 'render_ui'
if config.tool_response or is_ui_resource or is_ask_user or is_render_ui:
new_content = f"[{message_tag}] {msg.name}\n{msg.text}\n"
# Collect full content

19
skills/common/data-dashboard/.claude-plugin/plugin.json Normal file
View File

@ -0,0 +1,19 @@
{
"name": "data-dashboard",
"description": "Renders data as an interactive dashboard card UI using the mcp-ui protocol.",
"hooks": {
"PrePrompt": [
{
"type": "command",
"command": "python hooks/pre_prompt.py"
}
]
},
"mcpServers": {
"data_dashboard": {
"transport": "stdio",
"command": "python",
"args": ["./dashboard_server.py", "{bot_id}"]
}
}
}

187
skills/common/data-dashboard/dashboard_server.py Normal file
View File

@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""
Data Dashboard MCP Server - renders metric data as an interactive dashboard.
Returns UIResource via the mcp-ui protocol so the frontend renders it as an iframe.
"""
import asyncio
import json
from typing import Any, Dict, List
from mcp_ui_server import create_ui_resource, UIMetadataKey
from mcp_common import (
create_error_response,
create_initialize_response,
create_ping_response,
create_tools_list_response,
load_tools_from_json,
handle_mcp_streaming,
)
def _build_dashboard_html(title: str, metrics: List[Dict[str, str]]) -> str:
"""Build a self-contained HTML dashboard from metrics data."""
cards_html = ""
for m in metrics:
label = _esc(m.get("label", ""))
value = _esc(m.get("value", ""))
change = m.get("change", "")
change_type = m.get("change_type", "neutral")
change_html = ""
if change:
color = "#16a34a" if change_type == "up" else "#dc2626" if change_type == "down" else "#6b7280"
arrow = "▲" if change_type == "up" else "▼" if change_type == "down" else "—"
change_html = f'<span class="change" style="color:{color}"><span class="arrow">{arrow}</span> {_esc(change)}</span>'
cards_html += f"""
<div class="card">
<div class="card-label">{label}</div>
<div class="card-value">{value}</div>
{f'<div class="card-change">{change_html}</div>' if change_html else ''}
</div>"""
return f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{_esc(title)}</title>
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; background: #f8fafc; padding: 24px; }}
h1 {{ font-size: 20px; font-weight: 600; color: #0f172a; margin-bottom: 20px; }}
.grid {{ display: grid; grid-template-columns: repeat(auto-fill, minmax(200px, 1fr)); gap: 16px; }}
.card {{ background: #fff; border: 1px solid #e2e8f0; border-radius: 12px; padding: 20px; transition: box-shadow 0.15s; }}
.card:hover {{ box-shadow: 0 4px 12px rgba(0,0,0,0.06); }}
.card-label {{ font-size: 13px; color: #64748b; margin-bottom: 8px; }}
.card-value {{ font-size: 28px; font-weight: 700; color: #0f172a; }}
.card-change {{ margin-top: 8px; font-size: 13px; font-weight: 500; }}
.arrow {{ font-size: 10px; }}
</style>
</head>
<body>
<h1>{_esc(title)}</h1>
<div class="grid">{cards_html}
</div>
</body>
</html>"""
def _esc(text: str) -> str:
"""Minimal HTML escaping."""
return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;").replace('"', "&quot;")
def render_dashboard(title: str, metrics: List[Dict[str, str]]) -> Dict[str, Any]:
"""Create a UIResource dashboard and serialize it as JSON for TOOL_RESPONSE."""
try:
html = _build_dashboard_html(title, metrics)
uri_slug = title.replace(" ", "-").lower()[:50]
ui_resource = create_ui_resource(
{
"uri": f"ui://data-dashboard/{uri_slug}",
"content": {"type": "rawHtml", "htmlString": html},
"encoding": "text",
"uiMetadata": {
UIMetadataKey.PREFERRED_FRAME_SIZE: ["100%", "auto"],
},
}
)
resource_json = json.dumps(
ui_resource.model_dump(mode="json"), ensure_ascii=False
)
return {"content": [{"type": "text", "text": resource_json}]}
except Exception as e:
return {
"content": [{"type": "text", "text": f"Error creating dashboard: {str(e)}"}]
}
async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]:
"""Handle an MCP request."""
try:
method = request.get("method")
params = request.get("params", {})
request_id = request.get("id")
if method == "initialize":
return create_initialize_response(request_id, "data-dashboard")
elif method == "ping":
return create_ping_response(request_id)
elif method == "tools/list":
tools = load_tools_from_json("dashboard_tools.json")
if not tools:
tools = [
{
"name": "render_dashboard",
"description": "Render a data dashboard with metric cards.",
"inputSchema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"metrics": {
"type": "array",
"items": {
"type": "object",
"properties": {
"label": {"type": "string"},
"value": {"type": "string"},
"change": {"type": "string"},
"change_type": {"type": "string", "enum": ["up", "down", "neutral"]},
},
"required": ["label", "value"],
},
},
},
"required": ["title", "metrics"],
},
}
]
return create_tools_list_response(request_id, tools)
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "render_dashboard":
title = arguments.get("title", "Dashboard")
metrics = arguments.get("metrics", [])
if not metrics:
return create_error_response(
request_id, -32602, "Missing required parameter: metrics"
)
result = render_dashboard(title, metrics)
return {"jsonrpc": "2.0", "id": request_id, "result": result}
else:
return create_error_response(
request_id, -32601, f"Unknown tool: {tool_name}"
)
else:
return create_error_response(
request_id, -32601, f"Unknown method: {method}"
)
except Exception as e:
return create_error_response(
request.get("id"), -32603, f"Internal error: {str(e)}"
)
async def main():
"""Main entry point."""
await handle_mcp_streaming(handle_request)
if __name__ == "__main__":
asyncio.run(main())

43
skills/common/data-dashboard/dashboard_tools.json Normal file
View File

@ -0,0 +1,43 @@
[
{
"name": "render_dashboard",
"description": "Render an interactive data dashboard with metric cards. Pass an array of metrics, each with label, value, and optional change/change_type fields. The dashboard is rendered as an HTML card UI.",
"inputSchema": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Dashboard title, e.g. 'Sales Overview'"
},
"metrics": {
"type": "array",
"description": "Array of metric objects to display as cards",
"items": {
"type": "object",
"properties": {
"label": {
"type": "string",
"description": "Metric name, e.g. 'Revenue'"
},
"value": {
"type": "string",
"description": "Metric value, e.g. '$12,345'"
},
"change": {
"type": "string",
"description": "Change indicator, e.g. '+12.5%' or '-3.2%'"
},
"change_type": {
"type": "string",
"enum": ["up", "down", "neutral"],
"description": "Direction of change"
}
},
"required": ["label", "value"]
}
}
},
"required": ["title", "metrics"]
}
}
]

13
skills/common/data-dashboard/hooks/dashboard_guide.md Normal file
View File

@ -0,0 +1,13 @@
## `render_dashboard` usage guide
### When to use render_dashboard:
- When the user wants to visualize data as metric cards (KPIs, stats, numbers)
- When summarizing numerical results (sales, traffic, performance metrics)
- When comparing multiple data points side by side
### How to call:
render_dashboard(title="Sales Overview", metrics=[
{"label": "Revenue", "value": "$12,345", "change": "+12.5%", "change_type": "up"},
{"label": "Users", "value": "1,234", "change": "-3.2%", "change_type": "down"},
{"label": "Conversion", "value": "4.5%", "change_type": "neutral"}
])

8
skills/common/data-dashboard/hooks/pre_prompt.py Normal file
View File

@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""PrePrompt hook - inject dashboard tool guide into system prompt."""
import os
guide_path = os.path.join(os.path.dirname(__file__), "dashboard_guide.md")
if os.path.exists(guide_path):
with open(guide_path, "r", encoding="utf-8") as f:
print(f.read())

252
skills/common/data-dashboard/mcp_common.py Normal file
View File

@ -0,0 +1,252 @@
#!/usr/bin/env python3
"""
Shared utility functions for the MCP server.
Provides common functionality for path handling, file validation, and request processing.
"""
import json
import os
import sys
import asyncio
from typing import Any, Dict, List, Optional, Union
import re
def get_allowed_directory():
"""Get the directory that is allowed to be accessed."""
# Prefer dataset_dir passed through command-line arguments.
if len(sys.argv) > 1:
dataset_dir = sys.argv[1]
return os.path.abspath(dataset_dir)
# Read the project data directory from the environment variable.
project_dir = os.getenv("PROJECT_DATA_DIR", "./projects/data")
return os.path.abspath(project_dir)
def resolve_file_path(file_path: str, default_subfolder: str = "default") -> str:
"""
Resolve a file path, supporting both folder/document.txt and document.txt formats.
Args:
file_path: Input file path.
default_subfolder: Default subfolder name to use when only a filename is provided.
Returns:
The resolved full file path.
"""
# If the path contains a folder separator, use it directly.
if '/' in file_path or '\\' in file_path:
clean_path = file_path.replace('\\', '/')
# Remove the projects/ prefix if it exists.
if clean_path.startswith('projects/'):
clean_path = clean_path[9:] # Remove the 'projects/' prefix.
elif clean_path.startswith('./projects/'):
clean_path = clean_path[11:] # Remove the './projects/' prefix.
else:
# If only a filename is provided, add the default subfolder.
clean_path = f"{default_subfolder}/{file_path}"
# Get the allowed directory.
project_data_dir = get_allowed_directory()
# Try to locate the file directly under the project directory.
full_path = os.path.join(project_data_dir, clean_path.lstrip('./'))
if os.path.exists(full_path):
return full_path
# If the direct path does not exist, try a recursive search.
found = find_file_in_project(clean_path, project_data_dir)
if found:
return found
# If this is a bare filename and it was not found under the default subfolder,
# try looking in the project root.
if '/' not in file_path and '\\' not in file_path:
root_path = os.path.join(project_data_dir, file_path)
if os.path.exists(root_path):
return root_path
raise FileNotFoundError(f"File not found: {file_path} (searched in {project_data_dir})")
def find_file_in_project(filename: str, project_dir: str) -> Optional[str]:
"""Recursively search for a file inside the project directory."""
# If filename includes a path, only search within the specified path.
if '/' in filename:
parts = filename.split('/')
target_file = parts[-1]
search_dir = os.path.join(project_dir, *parts[:-1])
if os.path.exists(search_dir):
target_path = os.path.join(search_dir, target_file)
if os.path.exists(target_path):
return target_path
else:
# For a bare filename, recursively search the whole project directory.
for root, dirs, files in os.walk(project_dir):
if filename in files:
return os.path.join(root, filename)
return None
def load_tools_from_json(tools_file_name: str) -> List[Dict[str, Any]]:
"""Load tool definitions from a JSON file."""
try:
tools_file = os.path.join(os.path.dirname(__file__), tools_file_name)
if os.path.exists(tools_file):
with open(tools_file, 'r', encoding='utf-8') as f:
return json.load(f)
else:
# If the JSON file does not exist, use the default definitions.
return []
except Exception as e:
print(f"Warning: Unable to load tool definition JSON file: {str(e)}")
return []
def create_error_response(request_id: Any, code: int, message: str) -> Dict[str, Any]:
"""Create a standardized error response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": code,
"message": message
}
}
def create_success_response(request_id: Any, result: Any) -> Dict[str, Any]:
"""Create a standardized success response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
def create_initialize_response(request_id: Any, server_name: str, server_version: str = "1.0.0") -> Dict[str, Any]:
"""Create a standardized initialize response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": server_name,
"version": server_version
}
}
}
def create_ping_response(request_id: Any) -> Dict[str, Any]:
"""Create a standardized ping response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"pong": True
}
}
def create_tools_list_response(request_id: Any, tools: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Create a standardized tools/list response."""
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": tools
}
}
def is_regex_pattern(pattern: str) -> bool:
"""Check whether a string should be treated as a regular expression pattern."""
# Check the /pattern/ format.
if pattern.startswith('/') and pattern.endswith('/') and len(pattern) > 2:
return True
# Check the r"pattern" or r'pattern' format.
if pattern.startswith(('r"', "r'")) and pattern.endswith(('"', "'")) and len(pattern) > 3:
return True
# Check whether it contains regex metacharacters.
regex_chars = {'*', '+', '?', '|', '(', ')', '[', ']', '{', '}', '^', '$', '\\', '.'}
return any(char in pattern for char in regex_chars)
def compile_pattern(pattern: str) -> Union[re.Pattern, str, None]:
"""Compile a regex pattern, or return the original string if it is not regex."""
if not is_regex_pattern(pattern):
return pattern
try:
# Handle the /pattern/ format.
if pattern.startswith('/') and pattern.endswith('/'):
regex_body = pattern[1:-1]
return re.compile(regex_body)
# Handle the r"pattern" or r'pattern' format.
if pattern.startswith(('r"', "r'")) and pattern.endswith(('"', "'")):
regex_body = pattern[2:-1]
return re.compile(regex_body)
# Directly compile strings that contain regex metacharacters.
return re.compile(pattern)
except re.error as e:
# If compilation fails, return None to indicate an invalid regex.
print(f"Warning: Regular expression '{pattern}' compilation failed: {e}")
return None
async def handle_mcp_streaming(request_handler):
"""Handle the standard main loop for MCP requests."""
try:
while True:
# Read from stdin
line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
if not line:
break
line = line.strip()
if not line:
continue
try:
request = json.loads(line)
response = await request_handler(request)
# Write to stdout
sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
sys.stdout.flush()
except json.JSONDecodeError:
error_response = {
"jsonrpc": "2.0",
"error": {
"code": -32700,
"message": "Parse error"
}
}
sys.stdout.write(json.dumps(error_response, ensure_ascii=False) + "\n")
sys.stdout.flush()
except Exception as e:
error_response = {
"jsonrpc": "2.0",
"error": {
"code": -32603,
"message": f"Internal error: {str(e)}"
}
}
sys.stdout.write(json.dumps(error_response, ensure_ascii=False) + "\n")
sys.stdout.flush()
except KeyboardInterrupt:
pass

4
skills/common/mcp-ui/mcp_ui_tools.json
View File

@ -24,8 +24,8 @@
},
"height": {
"type": "string",
"description": "CSS height for the iframe. Default: '400px'",
"default": "400px"
"description": "CSS height for the iframe. Set to a fixed value (e.g. '300px', '600px') matching the HTML content height. Use 'auto' only when the HTML is responsive and has no fixed height. Default: 'auto'",
"default": "auto"
}
},
"required": ["title", "html_content"]

64
skills/common/mcp-ui/ui_render_server.py
View File

@ -1,7 +1,6 @@
#!/usr/bin/env python3
"""
MCP UI Server - provides interactive UI rendering tools.
Uses mcp-ui-server SDK to create standard UIResource objects.
"""
import asyncio
@ -9,8 +8,6 @@ import json
import sys
from typing import Any, Dict
from mcp_ui_server import create_ui_resource, UIMetadataKey
from mcp_common import (
create_error_response,
create_initialize_response,
@ -39,52 +36,22 @@ def ask_user() -> Dict[str, Any]:
}
def render_ui(
title: str,
html_content: str = "",
url: str = "",
width: str = "100%",
height: str = "400px",
) -> Dict[str, Any]:
"""Create a UI resource and serialize it as JSON text for passthrough.
RENDER_UI_RESPONSE = "UI rendered."
Supports two modes:
- rawHtml: provide html_content to render custom HTML.
- externalUrl: provide url to embed an external webpage.
The UIResource is serialized as a JSON string inside a type:"text" content block.
def render_ui() -> Dict[str, Any]:
"""Return a minimal fixed response for render_ui tool.
The actual html_content/url is already in the TOOL_CALL arguments,
so the frontend parses them directly from there. This response only
serves to acknowledge the tool call and minimize token usage in the
subsequent LLM inference round.
"""
try:
uri_slug = title.replace(" ", "-").lower()[:50]
if url:
content = {"type": "externalUrl", "iframeUrl": url}
else:
content = {"type": "rawHtml", "htmlString": html_content}
ui_resource = create_ui_resource(
{
"uri": f"ui://mcp-ui-skill/{uri_slug}",
"content": content,
"encoding": "text",
"uiMetadata": {
UIMetadataKey.PREFERRED_FRAME_SIZE: [width, height],
},
}
)
resource_json = json.dumps(
ui_resource.model_dump(mode="json"), ensure_ascii=False
)
return {"content": [{"type": "text", "text": resource_json}]}
except Exception as e:
return {
"content": [
{"type": "text", "text": f"Error creating UI resource: {str(e)}"}
]
}
return {
"content": [
{"type": "text", "text": RENDER_UI_RESPONSE}
]
}
async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]:
@ -134,18 +101,15 @@ async def handle_request(request: Dict[str, Any]) -> Dict[str, Any]:
arguments = params.get("arguments", {})
if tool_name == "render_ui":
title = arguments.get("title", "UI Widget")
html_content = arguments.get("html_content", "")
url = arguments.get("url", "")
width = arguments.get("width", "100%")
height = arguments.get("height", "400px")
if not html_content and not url:
return create_error_response(
request_id, -32602, "Missing required parameter: html_content or url"
)
result = render_ui(title, html_content, url, width, height)
result = render_ui()
return {"jsonrpc": "2.0", "id": request_id, "result": result}
elif tool_name == "ask_user":