"""
Agnes 2.0 Flash MCP 视觉服务器
提供多工具视觉能力：看图问答、裁剪、取色、像素对比、抠图、定位
"""
import base64
import json
import mimetypes
import os
import sys
import uuid
from collections import Counter
from io import BytesIO
from pathlib import Path
import numpy as np
import requests
from PIL import Image, ImageDraw, ImageFilter
# ============================================================
# 配置
# ============================================================
def _get_config():
"""获取 Agnes API 配置"""
api_key = "sk-m4q4S2YPpVzMROKLh5DhK2kK3Y54qoI8DbJ2QOv9MBeIJ4qA"
return {
"api_key": api_key,
"base_url": "https://api.agnes-ai.cn/v1",
"model": "agnes-2.0-flash",
}
# 输出目录——存放裁剪/热力图等生成图片
OUTPUT_DIR = Path(__file__).parent / "output"
OUTPUT_DIR.mkdir(exist_ok=True)
# ============================================================
# 工具定义
# ============================================================
TOOLS = [
{
"name": "describe_image",
"description": "使用 Agnes 多模态模型识别/描述图片内容。支持单图问答、多图对比、JSON 结构化输出",
"inputSchema": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "单张图片路径（与 image_paths 二选一）",
},
"image_paths": {
"type": "array",
"items": {"type": "string"},
"description": "多张图片路径列表（与 image_path 二选一）",
},
"prompt": {
"type": "string",
"description": "对图片的提问或描述要求",
"default": "请详细描述这张图片的内容",
},
"json_mode": {
"type": "boolean",
"description": "是否以 JSON 格式返回结构化结果",
"default": False,
},
},
},
},
{
"name": "vision_crop",
"description": "按像素坐标裁剪图片指定区域，并可选放大倍数",
"inputSchema": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "图片文件路径",
},
"region": {
"type": "string",
"description": "裁剪区域，格式 'x1,y1,x2,y2'，如 '100,200,300,400'",
},
"scale": {
"type": "number",
"description": "放大倍数，默认 2.0",
"default": 2.0,
},
},
"required": ["image_path", "region"],
},
},
{
"name": "vision_colors",
"description": "提取图片主色调，返回十六进制色值和占比",
"inputSchema": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "图片文件路径",
},
"top_n": {
"type": "integer",
"description": "返回前 N 个主色，默认 5",
"default": 5,
},
},
"required": ["image_path"],
},
},
{
"name": "vision_pixel_diff",
"description": "逐像素对比两张图片，输出差异率、差异像素数，并生成红色热力图标注差异区域",
"inputSchema": {
"type": "object",
"properties": {
"original_path": {
"type": "string",
"description": "原图/基准图路径",
},
"modified_path": {
"type": "string",
"description": "修改后/对比图路径",
},
"threshold": {
"type": "integer",
"description": "通道差异阈值（0-255），默认 16",
"default": 16,
},
},
"required": ["original_path", "modified_path"],
},
},
{
"name": "vision_extract_foreground",
"description": "从纯色/简单背景中抠出前景物体，返回透明背景 PNG",
"inputSchema": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "图片文件路径",
},
"tolerance": {
"type": "integer",
"description": "颜色容差（0-255），默认 30",
"default": 30,
},
},
"required": ["image_path"],
},
},
{
"name": "vision_ground",
"description": "在图片中定位目标元素，返回其像素坐标框 (x1,y1,x2,y2)",
"inputSchema": {
"type": "object",
"properties": {
"image_path": {
"type": "string",
"description": "图片文件路径",
},
"target": {
"type": "string",
"description": "要定位的目标描述，如'发送按钮'、'搜索输入框'",
},
},
"required": ["image_path", "target"],
},
},
]
# ============================================================
# 通用工具函数
# ============================================================
def encode_image(image_path: str) -> str:
"""将图片编码为 base64 data URL"""
path = Path(image_path)
if not path.exists():
raise FileNotFoundError(f"图片文件不存在: {image_path}")
mime_type, _ = mimetypes.guess_type(str(path))
if mime_type is None or not mime_type.startswith("image/"):
mime_type = "image/png"
data = path.read_bytes()
b64 = base64.b64encode(data).decode("ascii")
return f"data:{mime_type};base64,{b64}"
def image_to_base64(img: Image.Image, fmt: str = "PNG") -> str:
"""将 Pillow Image 对象转为 base64 字符串"""
buf = BytesIO()
img.save(buf, format=fmt)
return base64.b64encode(buf.getvalue()).decode("ascii")
def clean_surrogates(text: str) -> str:
"""移除非法代理项字符"""
if not text:
return text
return text.encode("utf-8", errors="replace").decode("utf-8")
def save_output_image(img: Image.Image, prefix: str = "img") -> Path:
"""保存图片到输出目录，返回路径"""
name = f"{prefix}_{uuid.uuid4().hex[:8]}.png"
path = OUTPUT_DIR / name
img.save(path, "PNG")
return path
# ============================================================
# Agnes API 调用
# ============================================================
def call_agnes(image_paths: list[str], prompt: str, json_mode: bool = False) -> str:
"""调用 Agnes 2.0 Flash 进行图片识别，支持多图"""
config = _get_config()
content = [{"type": "text", "text": prompt}]
for path in image_paths:
data_url = encode_image(path)
content.append({"type": "image_url", "image_url": {"url": data_url}})
if json_mode:
content.insert(0, {
"type": "text",
"text": "请以 JSON 格式输出结果，包含字段：summary（摘要）、details（详细描述）。"
})
body = {
"model": config["model"],
"max_tokens": 4096,
"messages": [{"role": "user", "content": content}],
}
if json_mode:
body["response_format"] = {"type": "json_object"}
try:
resp = requests.post(
f"{config['base_url']}/chat/completions",
headers={
"Authorization": f"Bearer {config['api_key']}",
"Content-Type": "application/json",
},
json=body,
timeout=120,
)
data = resp.json()
if "error" in data:
return f"Agnes API 错误: {data['error'].get('message', data['error'])}"
content = (
data.get("choices", [{}])[0]
.get("message", {})
.get("content", "")
)
if not content:
return f"API 返回空内容。原始响应: {resp.text[:300]}"
return clean_surrogates(content)
except requests.Timeout:
return "Agnes API 请求超时（120s）"
except requests.RequestException as e:
return f"Agnes API 请求失败: {e}"
except Exception as e:
return f"Agnes API 调用失败: {e}"
# ============================================================
# 工具实现
# ============================================================
def tool_describe_image(image_path: str = None, image_paths: list[str] = None,
prompt: str = "请详细描述这张图片的内容",
json_mode: bool = False) -> dict:
"""增强版 describe_image：支持单图/多图/JSON 模式"""
paths = []
if image_paths:
paths = image_paths
elif image_path:
paths = [image_path]
else:
return {"text": "请提供 image_path 或 image_paths 参数"}
# 验证所有文件存在
missing = [p for p in paths if not Path(p).exists()]
if missing:
return {"text": f"以下图片文件不存在: {', '.join(missing)}"}
result = call_agnes(paths, prompt, json_mode=json_mode)
return {"text": result}
def tool_vision_crop(image_path: str, region: str, scale: float = 2.0) -> dict:
"""按坐标裁剪图片并放大"""
path = Path(image_path)
if not path.exists():
return {"text": f"图片文件不存在: {image_path}"}
try:
parts = [int(x.strip()) for x in region.split(",")]
if len(parts) != 4:
return {"text": f"region 格式错误，应为 'x1,y1,x2,y2'，收到: {region}"}
x1, y1, x2, y2 = parts
except ValueError:
return {"text": f"region 参数必须为整数，收到: {region}"}
if x1 >= x2 or y1 >= y2:
return {"text": f"无效的裁剪区域: x1={x1} >= x2={x2} 或 y1={y1} >= y2={y2}"}
try:
img = Image.open(path)
cropped = img.crop((x1, y1, x2, y2))
if scale != 1.0:
w, h = cropped.size
cropped = cropped.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
# 保存输出并返回 base64
out_path = save_output_image(cropped, "crop")
b64 = image_to_base64(cropped)
return {
"text": (
f"裁剪完成\n"
f" 区域: ({x1},{y1}) → ({x2},{y2})\n"
f" 原始尺寸: {x2 - x1} × {y2 - y1} px\n"
f" 放大后: {cropped.size[0]} × {cropped.size[1]} px\n"
f" 保存到: {out_path}"
),
"image": b64,
}
except Exception as e:
return {"text": f"裁剪失败: {e}"}
def tool_vision_colors(image_path: str, top_n: int = 5) -> dict:
"""提取图片主色调"""
path = Path(image_path)
if not path.exists():
return {"text": f"图片文件不存在: {image_path}"}
top_n = max(1, min(20, top_n))
try:
img = Image.open(path).convert("RGB")
# 缩小加速
w, h = img.size
if w * h > 200 * 200:
ratio = min(200 / w, 200 / h)
img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS)
# 量化压缩颜色数
quantized = img.quantize(colors=32, method=Image.Quantize.MEDIANCUT)
palette = quantized.getpalette()
color_counts = quantized.getcolors()
if not color_counts:
return {"text": "无法提取颜色信息"}
# 按出现次数排序
color_counts.sort(key=lambda x: x[0], reverse=True)
total = sum(c[0] for c in color_counts)
colors = []
for count, idx in color_counts[:top_n]:
r = palette[idx * 3]
g = palette[idx * 3 + 1]
b = palette[idx * 3 + 2]
hex_color = f"#{r:02x}{g:02x}{b:02x}"
pct = round(count / total * 100, 1)
colors.append({"hex": hex_color, "rgb": (r, g, b), "percent": pct})
text_lines = [f"主色调提取（前 {top_n}/{len(color_counts)} 色）:"]
for i, c in enumerate(colors, 1):
text_lines.append(f" {i}. {c['hex']} RGB{c['rgb']} {c['percent']}%")
return {"text": "\n".join(text_lines)}
except Exception as e:
return {"text": f"取色失败: {e}"}
def tool_vision_pixel_diff(original_path: str, modified_path: str,
threshold: int = 16) -> dict:
"""逐像素对比两张图片"""
orig = Path(original_path)
mod = Path(modified_path)
if not orig.exists():
return {"text": f"原图不存在: {original_path}"}
if not mod.exists():
return {"text": f"对比图不存在: {modified_path}"}
threshold = max(0, min(255, threshold))
try:
img_a = Image.open(orig).convert("RGB")
img_b = Image.open(mod).convert("RGB")
# 统一尺寸（以原图为准）
if img_a.size != img_b.size:
img_b = img_b.resize(img_a.size, Image.LANCZOS)
arr_a = np.array(img_a, dtype=np.int16)
arr_b = np.array(img_b, dtype=np.int16)
# 逐通道差异
diff = np.abs(arr_a - arr_b)
max_diff = int(diff.max())
# 任一通道差异超过 threshold 即为差异像素
channel_max = diff.max(axis=2)
diff_mask = channel_max > threshold
diff_pixels = int(diff_mask.sum())
total_pixels = diff_mask.size
diff_pct = round(diff_pixels / total_pixels * 100, 2)
# 生成热力图：差异区域标红
heat = np.array(img_a, dtype=np.uint8)
# 差异像素 → 红色叠加
red_overlay = np.zeros_like(heat)
red_overlay[diff_mask] = [255, 0, 0]
# 混合：原图 + 半透明红色
blend = np.where(diff_mask[:, :, None], (heat * 0.5 + red_overlay * 0.5).astype(np.uint8), heat)
heat_img = Image.fromarray(blend, "RGB")
out_path = save_output_image(heat_img, "diff")
b64 = image_to_base64(heat_img)
# 最差区域分析（8×8 网格）
h, w = diff_mask.shape
grid_h, grid_w = max(1, h // 8), max(1, w // 8)
worst_regions = []
for gy in range(8):
for gx in range(8):
y0, y1 = gy * grid_h, (gy + 1) * grid_h if gy < 7 else h
x0, x1 = gx * grid_w, (gx + 1) * grid_w if gx < 7 else w
chunk = diff_mask[y0:y1, x0:x1]
rate = int(chunk.sum()) / chunk.size
worst_regions.append((rate, gx, gy, x0, y0, x1, y1))
worst_regions.sort(key=lambda r: r[0], reverse=True)
top3 = worst_regions[:3]
lines = [
f"像素对比完成",
f" 总像素: {total_pixels:,}",
f" 差异像素: {diff_pixels:,} ({diff_pct}%)",
f" 通道阈值: {threshold}/255",
f" 最大通道差异: {max_diff}",
f" 热力图: {out_path}",
"",
f"差异最集中的 8×8 网格区域（前 3）:",
]
for i, (rate, gx, gy, x0, y0, x1, y1) in enumerate(top3, 1):
lines.append(f" {i}. 网格({gx},{gy}) ({x0},{y0})~({x1},{y1}) 差异率 {rate*100:.1f}%")
return {
"text": "\n".join(lines),
"image": b64,
}
except Exception as e:
return {"text": f"像素对比失败: {e}"}
def tool_vision_extract_foreground(image_path: str, tolerance: int = 30) -> dict:
"""从纯色背景中抠出前景物体"""
path = Path(image_path)
if not path.exists():
return {"text": f"图片文件不存在: {image_path}"}
tolerance = max(0, min(255, tolerance))
try:
img = Image.open(path).convert("RGBA")
pixels = img.load()
w, h = img.size
# 从四角采样背景色
corners = [
pixels[0, 0],
pixels[w - 1, 0],
pixels[0, h - 1],
pixels[w - 1, h - 1],
]
# 取出现最多的颜色作为背景色
bg_color = Counter(corners).most_common(1)[0][0]
# 标记背景像素
bg_r, bg_g, bg_b, bg_a = bg_color
mask = Image.new("L", (w, h), 255)
mask_pixels = mask.load()
for y in range(h):
for x in range(w):
pr, pg, pb, pa = pixels[x, y]
if (abs(pr - bg_r) <= tolerance and
abs(pg - bg_g) <= tolerance and
abs(pb - bg_b) <= tolerance):
mask_pixels[x, y] = 0 # 背景 → 透明
# 模糊边缘平滑
mask = mask.filter(ImageFilter.SMOOTH)
# 应用遮罩
result = Image.new("RGBA", (w, h), (0, 0, 0, 0))
result.paste(img, (0, 0), mask)
out_path = save_output_image(result, "foreground")
b64 = image_to_base64(result)
return {
"text": (
f"抠图完成\n"
f" 背景色: RGB({bg_r},{bg_g},{bg_b}) 容差: {tolerance}\n"
f" 图片尺寸: {w} × {h}\n"
f" 保存到: {out_path}"
),
"image": b64,
}
except Exception as e:
return {"text": f"抠图失败: {e}"}
def tool_vision_ground(image_path: str, target: str) -> dict:
"""在图片中定位目标元素，返回坐标"""
path = Path(image_path)
if not path.exists():
return {"text": f"图片文件不存在: {image_path}"}
# 获取图片尺寸
try:
with Image.open(path) as img:
img_w, img_h = img.size
except Exception as e:
return {"text": f"无法读取图片: {e}"}
prompt = (
f"请在图片中精确找到「{target}」的位置。\n"
f"图片尺寸: {img_w}×{img_h} px。\n"
f"请以如下格式返回坐标:\n"
f"x1=<数字>, y1=<数字>, x2=<数字>, y2=<数字>\n"
f"坐标范围: x 0~{img_w}, y 0~{img_h}\n"
f"如果找不到，请说明原因。"
)
result = call_agnes([image_path], prompt)
# 尝试从结果中解析坐标
import re
coords = re.findall(r'x1=(\d+).*?y1=(\d+).*?x2=(\d+).*?y2=(\d+)', result, re.DOTALL)
if coords:
x1, y1, x2, y2 = map(int, coords[0])
return {
"text": (
f"定位完成: 「{target}」\n"
f" 坐标: ({x1}, {y1}) → ({x2}, {y2})\n"
f" 尺寸: {x2 - x1} × {y2 - y1} px\n"
f" 图片: {img_w} × {img_h} px\n"
f"\n原始回答:\n{result}"
),
}
# 没解析出坐标，返回原始结果
return {"text": f"未能解析坐标。模型返回:\n{result}"}
# ============================================================
# JSON-RPC 处理
# ============================================================
def send_json(data: dict) -> None:
"""发送 JSON-RPC 响应"""
try:
raw = json.dumps(data, ensure_ascii=False)
sys.stdout.buffer.write(raw.encode("utf-8", errors="replace") + b"\n")
except Exception:
raw = json.dumps(data, ensure_ascii=True)
sys.stdout.buffer.write(raw.encode("ascii") + b"\n")
sys.stdout.flush()
# 工具名称到实现函数的映射
TOOL_HANDLERS = {
"describe_image": tool_describe_image,
"vision_crop": tool_vision_crop,
"vision_colors": tool_vision_colors,
"vision_pixel_diff": tool_vision_pixel_diff,
"vision_extract_foreground": tool_vision_extract_foreground,
"vision_ground": tool_vision_ground,
}
def _build_content(tool_result: dict) -> list:
"""将工具结果格式化为 MCP content 数组"""
content = [{"type": "text", "text": tool_result.get("text", "")}]
if "image" in tool_result:
content.append({
"type": "image",
"data": tool_result["image"],
"mimeType": "image/png",
})
return content
def handle_request(msg: dict) -> None:
"""处理 JSON-RPC 请求"""
msg_id = msg.get("id")
method = msg.get("method", "")
params = msg.get("params", {})
if method == "initialize":
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {
"name": "agnes-vision",
"version": "2.0.0",
},
},
})
return
if method == "tools/list":
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"result": {"tools": TOOLS},
})
return
if method == "tools/call":
tool_name = params.get("name", "")
arguments = params.get("arguments", {})
handler = TOOL_HANDLERS.get(tool_name)
if handler is None:
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"error": {"code": -32601, "message": f"未知工具: {tool_name}"},
})
return
try:
result = handler(**arguments)
content = _build_content(result)
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"result": {"content": content},
})
except FileNotFoundError as e:
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"content": [{"type": "text", "text": str(e)}]
},
})
except Exception as e:
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"content": [{"type": "text", "text": f"工具执行错误: {e}"}]
},
})
return
if msg_id is None:
return
send_json({
"jsonrpc": "2.0",
"id": msg_id,
"error": {"code": -32601, "message": f"未知方法: {method}"},
})
def main() -> None:
"""主循环：从 stdin 读取 JSON-RPC 请求"""
if hasattr(sys.stdin, "reconfigure"):
sys.stdin.reconfigure(encoding="utf-8")
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
handle_request(msg)
except json.JSONDecodeError:
pass
if __name__ == "__main__":
main()