PubChem Chemical Safety MCP Server

""" PubChem MCP Server 核心服务 """ import logging from typing import List, Optional, Dict, Any from .pubchem_client import PubChemClient from .cache_service import CacheService from ..models.schemas import ( CompoundInfo, SafetyInfo, ToxicityData, ErrorResponse ) logger = logging.getLogger(__name__) class PubChemService: """PubChem MCP 核心服务""" def __init__(self, cache_service: CacheService): self.cache_service = cache_service async def get_compound_info(self, name: str) -> CompoundInfo: """获取化合物基础信息""" # 检查缓存 cached_data = await self.cache_service.get_compound_info(name) if cached_data: return CompoundInfo(**cached_data) # 从PubChem获取数据 async with PubChemClient() as client: # 获取CID cid = await client.get_compound_cid(name) if not cid: return CompoundInfo(name=name) # 获取基础信息 compound_data = await client.get_compound_by_name(name) if "error" in compound_data: return CompoundInfo(name=name) # 解析数据 info = self._parse_compound_data(name, cid, compound_data) # 缓存结果 await self.cache_service.set_compound_info(name, info.model_dump()) return info async def get_safety_info(self, cid: int) -> SafetyInfo: """获取安全分类信息""" # 检查缓存 cached_data = await self.cache_service.get_safety_info(cid) if cached_data: return SafetyInfo(cid=cid, **cached_data) # 从PubChem获取数据 async with PubChemClient() as client: safety_data = await client.get_safety_info(cid) if "error" in safety_data: return SafetyInfo(cid=cid) # 缓存结果 await self.cache_service.set_safety_info(cid, safety_data) return SafetyInfo(cid=cid, **safety_data) async def get_toxicity_data(self, cid: int) -> ToxicityData: """获取毒性数据""" # 检查缓存 cached_data = await self.cache_service.get_toxicity_data(cid) if cached_data: return ToxicityData(cid=cid, **cached_data) # 从PubChem获取数据 async with PubChemClient() as client: toxicity_data = await client.get_toxicity_data(cid) if "error" in toxicity_data: return ToxicityData(cid=cid) # 缓存结果 await self.cache_service.set_toxicity_data(cid, toxicity_data) return ToxicityData(cid=cid, **toxicity_data) def _parse_compound_data(self, name: str, cid: int, data: Dict[str, Any]) -> CompoundInfo: """解析化合物数据""" try: properties = data.get("PropertyTable", {}).get("Properties", []) if not properties: return CompoundInfo(name=name, cid=cid) props = properties[0] return CompoundInfo( cid=cid, name=name, molecular_formula=props.get("MolecularFormula"), molecular_weight=props.get("MolecularWeight"), iupac_name=props.get("IUPACName"), smiles=props.get("IsomericSMILES"), inchi_key=props.get("InChIKey"), synonyms=[] # 需要额外API调用获取 ) except Exception as e: logger.error(f"Error parsing compound data: {e}") return CompoundInfo(name=name, cid=cid)