Fujitsu Social Digital Twin MCP Server

Instructions

Implementation Reference

• src/fujitsu_sdt_mcp/server.py:608-680 (handler)

  The handler function decorated with @mcp.tool(), which implements the core logic to parse a natural language description and generate a structured simulation configuration JSON based on keyword matching for different simulation types (traffic, escooter, road_pricing). It handles input validation, keyword-based config generation, and error handling.

  @mcp.tool()
  async def create_natural_language_simulation_config(description: str, ctx: Optional[Context] = None) -> Dict[str, Any]:
      """Converts a natural language description into a structured simulation configuration, 
      interpreting user requirements into technical parameters for traffic simulation."""
      try:
          if not description:
              return format_api_error(400, "Description required")
          
          config = {
              "simulationType": "unknown",
              "parameters": {}
          }
          
          description_lower = description.lower()
          
          if any(keyword in description_lower for keyword in ["traffic", "congestion", "road", "signal"]):
              config["simulationType"] = "traffic"
              
              regions = ["Tokyo", "Osaka", "Nagoya", "Fukuoka", "Sapporo", "Sendai", "Hiroshima", "Kyoto"]
              for region in regions:
                  if region.lower() in description_lower:
                      config["parameters"]["region"] = region
                      break
              
              if "morning" in description_lower or "rush hour" in description_lower:
                  config["parameters"]["timeRange"] = "morning_rush"
              elif "evening" in description_lower:
                  config["parameters"]["timeRange"] = "evening_rush"
              elif "daytime" in description_lower:
                  config["parameters"]["timeRange"] = "daytime"
              
          elif any(keyword in description_lower for keyword in ["scooter", "e-scooter", "electric"]):
              config["simulationType"] = "escooter"
              
              count_match = re.search(r'(\d+) scooters', description)
              if count_match:
                  config["parameters"]["scooterCount"] = int(count_match.group(1))
              
              if "demand" in description_lower:
                  config["parameters"]["deploymentStrategy"] = "demand_based"
              elif "grid" in description_lower:
                  config["parameters"]["deploymentStrategy"] = "grid_based"
              elif "transit" in description_lower:
                  config["parameters"]["deploymentStrategy"] = "transit_focused"
              
          elif any(keyword in description_lower for keyword in ["pricing", "toll", "congestion charge"]):
              config["simulationType"] = "road_pricing"
              
              if "city center" in description_lower:
                  config["parameters"]["pricingZone"] = "city_center"
              elif "wider area" in description_lower:
                  config["parameters"]["pricingZone"] = "wider_area"
              elif "major roads" in description_lower:
                  config["parameters"]["pricingZone"] = "major_roads"
              
              if "fixed" in description_lower:
                  config["parameters"]["priceModel"] = "fixed"
              elif "time variable" in description_lower:
                  config["parameters"]["priceModel"] = "time_variable"
              elif "congestion" in description_lower:
                  config["parameters"]["priceModel"] = "congestion_variable"
              
          else:
              config["simulationType"] = "generic"
              config["parameters"]["description"] = description
          
          current_time = datetime.now().strftime("%Y%m%d_%H%M%S")
          config["name"] = f"{config['simulationType']}_{current_time}"
          
          return config
      except Exception as e:
          logger.error(f"Config generation error: {e}")
          return format_api_error(500, str(e))