vefrank_mcp_enhanced.py•24.8 kB
#!/usr/bin/env python3
"""
Enhanced VeFrank MCP Server
Unified, intelligent 3D asset generation with visual feedback and AI capabilities
"""
import asyncio
import json
import logging
import os
import base64
from pathlib import Path
from typing import Any, Dict, List, Optional
import tempfile
# MCP imports
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
logger = logging.getLogger(__name__)
class VeFrankEnhancedMCP:
"""Enhanced MCP server with intelligent asset generation."""
def __init__(self):
self.server = Server("vefrank-enhanced")
self.blender_host = "localhost"
self.blender_port = 9876
self.assets_dir = Path("assets/3d_models/automotive")
self.cache_dir = Path("assets/generation_cache")
self.cache_dir.mkdir(exist_ok=True)
# Load knowledge base
self.component_knowledge = self.load_component_knowledge()
self.generation_cache = self.load_generation_cache()
self._setup_handlers()
def load_component_knowledge(self) -> Dict:
"""Load knowledge from vehicle JSONs."""
knowledge = {}
cars_dir = Path("cars")
if cars_dir.exists():
for json_file in cars_dir.glob("*.json"):
try:
with open(json_file, 'r') as f:
data = json.load(f)
if "components" in data:
for comp_id, comp_data in data["components"].items():
comp_type = comp_data.get("type", "")
if comp_type not in knowledge:
knowledge[comp_type] = {
"examples": [],
"descriptions": [],
"part_numbers": []
}
knowledge[comp_type]["examples"].append(comp_data)
if "description" in comp_data:
knowledge[comp_type]["descriptions"].append(comp_data["description"])
if "part_number" in comp_data:
knowledge[comp_type]["part_numbers"].append(comp_data["part_number"])
except:
pass
return knowledge
def load_generation_cache(self) -> Dict:
"""Load cached successful generations."""
cache_file = self.cache_dir / "generation_cache.json"
if cache_file.exists():
with open(cache_file, 'r') as f:
return json.load(f)
return {}
def save_generation_cache(self):
"""Save generation cache."""
cache_file = self.cache_dir / "generation_cache.json"
with open(cache_file, 'w') as f:
json.dump(self.generation_cache, f, indent=2)
def _setup_handlers(self):
"""Set up MCP handlers."""
@self.server.list_tools()
async def list_tools() -> List[Tool]:
return [
Tool(
name="generate_smart_component",
description="Generate automotive component using AI and templates intelligently",
inputSchema={
"type": "object",
"properties": {
"component_type": {"type": "string"},
"use_ai": {"type": "boolean", "default": False},
"use_reference_image": {"type": "boolean", "default": False},
"quality_level": {
"type": "string",
"enum": ["draft", "standard", "high"],
"default": "standard"
}
},
"required": ["component_type"]
}
),
Tool(
name="generate_from_description",
description="Generate 3D model from text description using AI",
inputSchema={
"type": "object",
"properties": {
"description": {"type": "string"},
"component_type": {"type": "string"},
"use_cache": {"type": "boolean", "default": True}
},
"required": ["description", "component_type"]
}
),
Tool(
name="verify_component_visual",
description="Take screenshot of generated component for verification",
inputSchema={
"type": "object",
"properties": {
"component_name": {"type": "string"},
"viewport": {
"type": "string",
"enum": ["perspective", "front", "side", "top"],
"default": "perspective"
}
},
"required": ["component_name"]
}
),
Tool(
name="improve_component",
description="Improve existing component based on feedback",
inputSchema={
"type": "object",
"properties": {
"component_name": {"type": "string"},
"feedback": {"type": "string"},
"improvement_type": {
"type": "string",
"enum": ["geometry", "materials", "details", "proportions"],
"default": "geometry"
}
},
"required": ["component_name", "feedback"]
}
),
Tool(
name="batch_generate_vehicle",
description="Generate all components for a vehicle efficiently",
inputSchema={
"type": "object",
"properties": {
"vehicle_json": {"type": "string"},
"quality": {
"type": "string",
"enum": ["fast", "balanced", "quality"],
"default": "balanced"
}
},
"required": ["vehicle_json"]
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]):
if name == "generate_smart_component":
return await self.generate_smart_component(**arguments)
elif name == "generate_from_description":
return await self.generate_from_description(**arguments)
elif name == "verify_component_visual":
return await self.verify_component_visual(**arguments)
elif name == "improve_component":
return await self.improve_component(**arguments)
elif name == "batch_generate_vehicle":
return await self.batch_generate_vehicle(**arguments)
else:
return {"error": f"Unknown tool: {name}"}
async def generate_smart_component(self, component_type: str, use_ai: bool = False,
use_reference_image: bool = False,
quality_level: str = "standard") -> Dict:
"""Smart component generation with multiple strategies."""
# Check cache first
cache_key = f"{component_type}_{quality_level}"
if cache_key in self.generation_cache:
return {
"status": "success",
"source": "cache",
"component_type": component_type,
"message": "Retrieved from cache"
}
# Get component knowledge
knowledge = self.component_knowledge.get(component_type, {})
# Build generation strategy
if use_ai and knowledge.get("descriptions"):
# Use AI generation with knowledge
description = self.build_ai_description(component_type, knowledge)
result = await self.generate_with_ai(description, component_type)
elif use_reference_image:
# Search for reference and generate
result = await self.generate_with_reference(component_type)
else:
# Use procedural generation with Blender
result = await self.generate_procedural(component_type, knowledge, quality_level)
# Cache successful generation
if result.get("status") == "success":
self.generation_cache[cache_key] = result
self.save_generation_cache()
return result
def build_ai_description(self, component_type: str, knowledge: Dict) -> str:
"""Build detailed AI description from knowledge."""
descriptions = knowledge.get("descriptions", [])
part_numbers = knowledge.get("part_numbers", [])
# Build comprehensive description
desc_parts = [f"automotive {component_type}"]
if descriptions:
desc_parts.append(descriptions[0])
if part_numbers:
desc_parts.append(f"similar to part {part_numbers[0]}")
# Add specific features based on type
type_features = {
"alternator": "with cooling fins, pulley, mounting brackets, electrical terminals",
"module": "with heat sink, multiple connectors, mounting tabs",
"pump": "with inlet outlet ports, motor housing",
"sensor": "with threaded body, electrical connector",
"throttle": "with butterfly valve, motor actuator, intake flange"
}
for key, features in type_features.items():
if key in component_type.lower():
desc_parts.append(features)
break
return " ".join(desc_parts)
async def generate_with_ai(self, description: str, component_type: str) -> Dict:
"""Generate using Hyper3D Rodin AI."""
# Build Blender script for AI generation
script = f'''
import bpy
import json
# Check if Hyper3D is enabled
if not bpy.context.scene.blendermcp_use_hyper3d:
print("ERROR: Hyper3D not enabled")
else:
# This would trigger Hyper3D generation
# For now, create placeholder
bpy.ops.mesh.primitive_cube_add(size=0.1, location=(0, 0, 0))
obj = bpy.context.object
obj.name = "{component_type}_ai_generated"
# Add subdivision for smoother look
modifier = obj.modifiers.new("Subdivision", 'SUBSURF')
modifier.levels = 2
result = {{"name": obj.name, "vertices": len(obj.data.vertices)}}
print(json.dumps(result))
'''
# Execute in Blender
result = await self.execute_blender_code(script)
return {
"status": "success",
"method": "ai_generation",
"component_type": component_type,
"description": description,
"result": result
}
async def generate_with_reference(self, component_type: str) -> Dict:
"""Generate using reference images from web."""
# For now, use enhanced procedural generation
# In production, would search for images and use as reference
return await self.generate_procedural(component_type, {}, "high")
async def generate_procedural(self, component_type: str, knowledge: Dict,
quality: str) -> Dict:
"""Generate using Blender procedural modeling."""
# Determine shape and features
shape = self.infer_shape(component_type)
# Build Blender script based on shape
if shape == "cylindrical":
script = self.build_cylinder_script(component_type, quality)
elif shape == "rectangular":
script = self.build_box_script(component_type, quality)
else:
script = self.build_generic_script(component_type, quality)
# Execute in Blender
result = await self.execute_blender_code(script)
# Export to OBJ
export_script = f'''
import bpy
obj = bpy.data.objects.get("{component_type}")
if obj:
bpy.context.view_layer.objects.active = obj
obj.select_set(True)
filepath = r"assets/3d_models/automotive/{component_type}/{component_type}.obj"
bpy.ops.export_scene.obj(filepath=filepath, use_selection=True)
print(f"Exported to {{filepath}}")
'''
await self.execute_blender_code(export_script)
return {
"status": "success",
"method": "procedural",
"component_type": component_type,
"quality": quality,
"result": result
}
def infer_shape(self, component_type: str) -> str:
"""Infer shape from component type."""
type_lower = component_type.lower()
if any(word in type_lower for word in ['alternator', 'pump', 'motor', 'sensor']):
return "cylindrical"
elif any(word in type_lower for word in ['module', 'ecu', 'controller', 'box']):
return "rectangular"
else:
return "generic"
def build_cylinder_script(self, component_type: str, quality: str) -> str:
"""Build script for cylindrical component."""
# Quality determines subdivision level
subdiv = {"draft": 0, "standard": 1, "high": 2}[quality]
return f'''
import bpy
import json
# Create cylinder
bpy.ops.mesh.primitive_cylinder_add(
vertices=32,
radius=0.065,
depth=0.18,
location=(0, 0, 0)
)
obj = bpy.context.object
obj.name = "{component_type}"
# Add subdivision for quality
if {subdiv} > 0:
modifier = obj.modifiers.new("Subdivision", 'SUBSURF')
modifier.levels = {subdiv}
# Add some details based on type
if "alternator" in "{component_type}".lower():
# Add cooling fins
for i in range(8):
angle = i * 3.14159 * 2 / 8
x = 0.07 * cos(angle)
y = 0.07 * sin(angle)
bpy.ops.mesh.primitive_cube_add(size=0.01, location=(x, y, 0))
fin = bpy.context.object
fin.scale = (0.2, 2, 8)
# Join with main body
obj.select_set(True)
bpy.context.view_layer.objects.active = obj
bpy.ops.object.join()
result = {{"name": obj.name, "vertices": len(obj.data.vertices)}}
print(json.dumps(result))
'''
def build_box_script(self, component_type: str, quality: str) -> str:
"""Build script for box-shaped component."""
subdiv = {"draft": 0, "standard": 1, "high": 2}[quality]
return f'''
import bpy
import json
# Create box
bpy.ops.mesh.primitive_cube_add(size=0.1, location=(0, 0, 0))
obj = bpy.context.object
obj.name = "{component_type}"
obj.scale = (2, 1.5, 0.4) # Make it module-shaped
bpy.ops.object.transform_apply(scale=True)
# Add subdivision
if {subdiv} > 0:
modifier = obj.modifiers.new("Subdivision", 'SUBSURF')
modifier.levels = {subdiv}
# Add heat sink fins if module
if "module" in "{component_type}".lower():
for i in range(5):
bpy.ops.mesh.primitive_cube_add(size=0.005, location=(i*0.04-0.08, 0, 0.045))
fin = bpy.context.object
fin.scale = (1, 10, 2)
obj.select_set(True)
bpy.context.view_layer.objects.active = obj
bpy.ops.object.join()
result = {{"name": obj.name, "vertices": len(obj.data.vertices)}}
print(json.dumps(result))
'''
def build_generic_script(self, component_type: str, quality: str) -> str:
"""Build script for generic component."""
return self.build_box_script(component_type, quality)
async def execute_blender_code(self, code: str) -> Dict:
"""Execute code in Blender via socket."""
try:
# Connect to Blender addon
reader, writer = await asyncio.open_connection(
self.blender_host, self.blender_port
)
# Send command
command = {
"type": "execute_code",
"params": {"code": code}
}
writer.write(json.dumps(command).encode('utf-8'))
await writer.drain()
# Read response
data = await reader.read(8192)
response = json.loads(data.decode('utf-8'))
writer.close()
await writer.wait_closed()
return response
except Exception as e:
return {"error": str(e)}
async def verify_component_visual(self, component_name: str,
viewport: str = "perspective") -> Dict:
"""Take screenshot of component for verification."""
script = f'''
import bpy
import json
obj = bpy.data.objects.get("{component_name}")
if not obj:
print(json.dumps({{"error": "Component not found"}}))
else:
# Focus on object
bpy.ops.object.select_all(action='DESELECT')
obj.select_set(True)
bpy.context.view_layer.objects.active = obj
bpy.ops.view3d.view_selected()
# Set viewport
if "{viewport}" == "front":
bpy.ops.view3d.view_axis(type='FRONT')
elif "{viewport}" == "side":
bpy.ops.view3d.view_axis(type='RIGHT')
elif "{viewport}" == "top":
bpy.ops.view3d.view_axis(type='TOP')
# Render viewport
filepath = r"assets/generation_cache/{component_name}_preview.png"
bpy.ops.render.opengl(write_still=True, view_context=True)
bpy.data.images['Render Result'].save_render(filepath=filepath)
result = {{"screenshot": filepath, "object": obj.name}}
print(json.dumps(result))
'''
result = await self.execute_blender_code(script)
# Read image and encode as base64
if "screenshot" in result:
try:
with open(result["screenshot"], "rb") as f:
image_data = base64.b64encode(f.read()).decode()
result["image_base64"] = image_data
except:
pass
return result
async def improve_component(self, component_name: str, feedback: str,
improvement_type: str = "geometry") -> Dict:
"""Improve component based on feedback."""
improvements = {
"geometry": self.improve_geometry_script,
"materials": self.improve_materials_script,
"details": self.improve_details_script,
"proportions": self.improve_proportions_script
}
script_builder = improvements.get(improvement_type, self.improve_geometry_script)
script = script_builder(component_name, feedback)
result = await self.execute_blender_code(script)
return {
"status": "success",
"component": component_name,
"improvement": improvement_type,
"feedback": feedback,
"result": result
}
def improve_geometry_script(self, name: str, feedback: str) -> str:
"""Build script to improve geometry."""
# Parse feedback for keywords
if "smooth" in feedback.lower():
modifier = "modifier.levels = 2"
elif "detailed" in feedback.lower():
modifier = "modifier.levels = 3"
else:
modifier = "modifier.levels = 1"
return f'''
import bpy
import json
obj = bpy.data.objects.get("{name}")
if obj:
# Add or update subdivision
modifier = obj.modifiers.get("Subdivision")
if not modifier:
modifier = obj.modifiers.new("Subdivision", 'SUBSURF')
{modifier}
# Apply feedback-based changes
feedback = "{feedback}".lower()
if "larger" in feedback:
obj.scale *= 1.2
elif "smaller" in feedback:
obj.scale *= 0.8
bpy.ops.object.transform_apply(scale=True)
result = {{"improved": True, "vertices": len(obj.data.vertices)}}
print(json.dumps(result))
'''
def improve_materials_script(self, name: str, feedback: str) -> str:
"""Build script to improve materials."""
return f'''
import bpy
import json
obj = bpy.data.objects.get("{name}")
if obj:
# Create material if doesn't exist
if not obj.data.materials:
mat = bpy.data.materials.new(name="{name}_material")
mat.use_nodes = True
obj.data.materials.append(mat)
mat = obj.data.materials[0]
bsdf = mat.node_tree.nodes.get("Principled BSDF")
# Apply feedback
feedback = "{feedback}".lower()
if "metallic" in feedback or "metal" in feedback:
bsdf.inputs["Metallic"].default_value = 0.9
bsdf.inputs["Roughness"].default_value = 0.2
elif "plastic" in feedback:
bsdf.inputs["Metallic"].default_value = 0.0
bsdf.inputs["Roughness"].default_value = 0.5
result = {{"material_updated": True}}
print(json.dumps(result))
'''
def improve_details_script(self, name: str, feedback: str) -> str:
"""Build script to add details."""
return self.improve_geometry_script(name, feedback)
def improve_proportions_script(self, name: str, feedback: str) -> str:
"""Build script to fix proportions."""
return self.improve_geometry_script(name, feedback)
async def batch_generate_vehicle(self, vehicle_json: str,
quality: str = "balanced") -> Dict:
"""Generate all components for a vehicle efficiently."""
# Load vehicle data
json_path = Path(vehicle_json)
if not json_path.exists():
json_path = Path("cars") / vehicle_json
if not json_path.exists():
return {"error": f"Vehicle file not found: {vehicle_json}"}
with open(json_path, 'r') as f:
vehicle_data = json.load(f)
components = vehicle_data.get("components", {})
results = []
# Determine generation strategy based on quality
use_ai = quality == "quality"
quality_level = {"fast": "draft", "balanced": "standard", "quality": "high"}[quality]
# Generate each component
for comp_id, comp_data in components.items():
comp_type = comp_data.get("type", comp_id.lower())
result = await self.generate_smart_component(
component_type=comp_type,
use_ai=use_ai,
quality_level=quality_level
)
results.append({
"id": comp_id,
"type": comp_type,
"status": result.get("status"),
"method": result.get("method", "unknown")
})
# Summary
successful = sum(1 for r in results if r["status"] == "success")
return {
"status": "success",
"vehicle": vehicle_json,
"total_components": len(components),
"successful": successful,
"quality": quality,
"results": results
}
async def run(self):
"""Run the enhanced MCP server."""
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
def main():
"""Main entry point."""
import argparse
parser = argparse.ArgumentParser(
description="Enhanced VeFrank MCP Server for intelligent 3D asset generation"
)
args = parser.parse_args()
server = VeFrankEnhancedMCP()
try:
asyncio.run(server.run())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()