PubChem Chemical Safety MCP Server

Instructions

Implementation Reference

• pubchem_mcp/mcp_server.py:73-110 (handler)

  The handler function get_safety_info, which is registered as an MCP tool, orchestrates the retrieval of safety data by checking the cache and then querying the PubChem client if necessary.

  async def get_safety_info(ctx: Context, cid: int) -> Dict[str, Any]:
      """
      获取GHS安全分类信息
      
      Args:
          cid: PubChem化合物ID
          
      Returns:
          包含信号词、GHS象形图、危害声明等安全信息的字典
      """
      try:
          # 初始化缓存服务
          cache_service = CacheService()
          await cache_service.initialize()
          
          # 检查缓存
          cached_data = await cache_service.get_safety_info(cid)
          if cached_data:
              await cache_service.close()
              return {"cid": cid, **cached_data}
          
          # 从PubChem获取数据
          async with PubChemClient() as client:
              safety_data = await client.get_safety_info(cid)
              
              if "error" in safety_data:
                  await cache_service.close()
                  return {"cid": cid, "error": safety_data["error"]}
              
              # 缓存结果
              await cache_service.set_safety_info(cid, safety_data)
              await cache_service.close()
              
              return {"cid": cid, **safety_data}
              
      except Exception as e:
          logger.error(f"Error getting safety info for CID {cid}: {e}")
          return {"cid": cid, "error": str(e)}

• pubchem_mcp/mcp_server.py:72-72 (registration)

  The get_safety_info tool is registered using the @app.tool() decorator.

  @app.tool()