Chemspace MCP Server

Instructions

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)

• src/chemspace_mcp/tokenmanager.py:9-62 (helper)

  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