Fujitsu Social Digital Twin MCP Server

Instructions

Implementation Reference

• src/fujitsu_sdt_mcp/server.py:488-502 (handler)

  The primary MCP tool handler for 'get_metrics', registered via @mcp.tool() decorator. Validates the simulation_id parameter and calls the API client to fetch metrics.

  @mcp.tool()
  async def get_metrics(simulation_id: str, ctx: Optional[Context] = None) -> Dict[str, Any]:
      """Retrieves comprehensive metrics and analytics data from a completed simulation, including 
      travel statistics, emissions data, and traffic flow information."""
      try:
          if not simulation_id:
              return format_api_error(400, "simulationId required")
          
          async with await get_http_client() as client:
              api_client = FujitsuSocialDigitalTwinClient(client)
              result = await api_client.get_metrics(simulation_id)
          return result
      except Exception as e:
          logger.error(f"Metrics retrieval error: {e}")
          return format_api_error(500, str(e))

• src/fujitsu_sdt_mcp/server.py:109-119 (helper)

  Core helper method in FujitsuSocialDigitalTwinClient that executes the HTTP GET request to the Fujitsu API endpoint /api/metrics/{simulation_id} and formats the response.

  async def get_metrics(self, simulation_id: str) -> Dict[str, Any]:
      try:
          response = await self.client.get(f"/api/metrics/{simulation_id}")
          response.raise_for_status()
          return format_simulation_result(response.json())
      except httpx.HTTPStatusError as e:
          logger.error(f"Metrics retrieval error: {e}")
          return format_api_error(e.response.status_code, str(e))
      except Exception as e:
          logger.error(f"Unexpected error retrieving metrics: {e}")
          return format_api_error(500, str(e))

• src/fujitsu_sdt_mcp/server.py:47-51 (helper)

  Helper function used by get_metrics to format successful API responses with success=True and data payload.

  def format_simulation_result(result: Dict[str, Any]) -> Dict[str, Any]:
      return {
          "success": True,
          "data": result
      }

• src/fujitsu_sdt_mcp/server.py:40-45 (helper)

  Helper function used by get_metrics to format error responses in a standard structure.

  def format_api_error(status_code: int, error_detail: str) -> Dict[str, Any]:
      return {
          "success": False,
          "status_code": status_code,
          "error": error_detail
      }

• src/fujitsu_sdt_mcp/server.py:178-200 (helper)

  MCP resource providing detailed explanations of the metrics returned by the get_metrics tool, serving as output schema documentation.

  @mcp.resource("resource://simulation_metrics_explanation")
  def get_simulation_metrics_explanation() -> str:
      """シミュレーションメトリクスの説明"""
      return """
  シミュレーションメトリクスの説明：
  
  1. co2: 総CO2排出量（kg）
  2. travelTime: 平均移動時間（分）
  3. travelCost: 平均移動コスト（円）
  4. trafficCountTotal: 総交通量
  5. escooterUsageRate: eスクーター利用率（%）
  6. escooterRevenue: eスクーター収益（円）
  7. escooterOpportunityLoss: eスクーター機会損失（円）
  8. bayCountWithLowBatteryScooters: バッテリー残量の少ないスクーターを含む充電ベイ数
  9. bayAvailableRate: ベイ利用可能率（%）
  10. toll: 通行料収入（円）
  
  交通カテゴリ：
  - car（車）, pedestrian（歩行者）, escooter（eスクーター）, bicycle（自転車）, truck（トラック）, lost（損失）, pt（公共交通）, park_ride（パーク&ライド）
  
  距離カテゴリ（km）：
  - car（車）, pedestrian（歩行者）, escooter（eスクーター）, bicycle（自転車）, bus（バス）, tram（路面電車）, subway（地下鉄）, rail（鉄道）
  """