PubChem Chemical Safety MCP Server

Instructions

Implementation Reference

• pubchem_mcp/mcp_server.py:113-151 (handler)

  The MCP tool handler for 'get_toxicity_data', which orchestrates the call to the cache service and PubChem client to retrieve toxicity data.

  @app.tool()
  async def get_toxicity_data(ctx: Context, cid: int) -> Dict[str, Any]:
      """
      获取毒性实验数据
      
      Args:
          cid: PubChem化合物ID
          
      Returns:
          包含急性毒性、生态毒性、致癌性等毒性数据的字典
      """
      try:
          # 初始化缓存服务
          cache_service = CacheService()
          await cache_service.initialize()
          
          # 检查缓存
          cached_data = await cache_service.get_toxicity_data(cid)
          if cached_data:
              await cache_service.close()
              return {"cid": cid, **cached_data}
          
          # 从PubChem获取数据
          async with PubChemClient() as client:
              toxicity_data = await client.get_toxicity_data(cid)
              
              if "error" in toxicity_data:
                  await cache_service.close()
                  return {"cid": cid, "error": toxicity_data["error"]}
              
              # 缓存结果
              await cache_service.set_toxicity_data(cid, toxicity_data)
              await cache_service.close()
              
              return {"cid": cid, **toxicity_data}
              
      except Exception as e:
          logger.error(f"Error getting toxicity data for CID {cid}: {e}")
          return {"cid": cid, "error": str(e)}