search_similarity
Find chemically similar compounds by SMILES string to identify building blocks and screening compounds for research and synthesis.
Instructions
Similarity search by SMILES
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| smiles | Yes | ||
| shipToCountry | No | The country you want your order to be shipped to as two-letter country ISO code, e.g DE, US, FR | US |
| count | No | Maximum number of results on a page | |
| page | No | Number of the page | |
| categories | No | A list of product categories to searchCSSB - In-stock building blocksCSSS - In-stock screening compoundsCSMB - Make-on-demand building blocksCSMS - Make-on-demand screening compoundsCSCS - Custom request |
Implementation Reference
- src/chemspace_mcp/tools.py:107-138 (handler)Handler function for the search_similarity tool. Makes a POST request to Chemspace API (uses substructure endpoint labeled as similarity). Uses token from ChemspaceTokenManager.@mcp.tool(enabled=True) async def search_similarity( smiles: str, shipToCountry: Country = "US", count: ResultCount = 10, page: ResultPage = 1, categories: ProductCategories = ["CSSB", "CSMB"], ): """Similarity search by SMILES""" access_token = await mgr.get_token() async with httpx.AsyncClient() as client: r = await client.post( url="https://api.chem-space.com/v4/search/sub", headers={ "Accept": "application/json; version=4.1", "Authorization": f"Bearer {access_token}", }, params={ "shipToCountry": shipToCountry, "count": count, "page": page, "categories": ",".join(categories), }, files={ "SMILES": (None, smiles), }, ) r.raise_for_status() data = r.json() return data
- src/chemspace_mcp/tools.py:8-36 (schema)Input schema definitions (Pydantic Annotated types) used for parameters in search_similarity and other search tools.Country = Annotated[ str, Field( description="The country you want your order to be shipped to as two-letter country ISO code, e.g DE, US, FR" ), ] ResultCount = Annotated[ int, Field(description="Maximum number of results on a page", ge=1) ] ResultPage = Annotated[int, Field(description="Number of the page", ge=1)] ProductCategory = Literal["CSSB", "CSSS", "CSMB", "CSMS", "CSCS"] ProductCategories = Annotated[ List[ProductCategory], Field( description=( "A list of product categories to search" "CSSB - In-stock building blocks" "CSSS - In-stock screening compounds" "CSMB - Make-on-demand building blocks" "CSMS - Make-on-demand screening compounds" "CSCS - Custom request" ), min_length=1, ), ]
- src/chemspace_mcp/__init__.py:5-10 (registration)Creates MCP instance and token manager, then calls register_tools to register the search_similarity tool (and others).mgr = ChemspaceTokenManager() mcp = FastMCP( "Chemspace MCP", instructions="Tools for retrieving synthesizable building blocks via the Chemspace API", ) register_tools(mcp, mgr)
- ChemspaceTokenManager class providing get_token() method used by search_similarity handler for authentication.class ChemspaceTokenManager: def __init__( self, api_key: str = os.environ.get("CHEMSPACE_API_KEY"), base_url: str = "https://api.chem-space.com/", ): self.api_key = api_key self.base_url = base_url self.auth_url = base_url + "auth/token" self.access_token = None self.expires_at = 0 self.token_cache = ( pathlib.Path(tempfile.gettempdir()) / ".token_cache" ).resolve() if self.token_cache.exists(): # read from cache with open(self.token_cache, "r") as fp: data = json.load(fp) self.access_token = data["access_token"] self.expires_at = float(data["expires_at"]) async def refresh_token(self): """Exchange API key for short-lived access token.""" async with httpx.AsyncClient() as client: r = await client.get( self.auth_url, headers={ "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", }, timeout=10, ) r.raise_for_status() data = r.json() self.access_token = data["access_token"] self.expires_at = time.time() + data["expires_in"] - 30 # refresh early # write to token cache with open(self.token_cache, "w") as fp: json.dump( {"access_token": self.access_token, "expires_at": self.expires_at}, fp, ) async def get_token(self): """Return a valid token, refreshing if needed.""" if time.time() >= self.expires_at: await self.refresh_token() return self.access_token