MCP Server Replicate

Instructions

Implementation Reference

• src/mcp_server_replicate/server.py:660-665 (handler)

  Main handler function for the list_collections tool, decorated with @mcp.tool() for registration. Calls ReplicateClient to fetch collections and converts to CollectionList.

  @mcp.tool()
  async def list_collections() -> CollectionList:
      """List available model collections on Replicate."""
      async with ReplicateClient(api_token=os.getenv("REPLICATE_API_TOKEN")) as client:
          result = await client.list_collections()
          return CollectionList(collections=[Collection(**collection) for collection in result])

• src/mcp_server_replicate/models/collection.py:16-19 (schema)

  Pydantic schema for the return type CollectionList used by the tool.

  class CollectionList(BaseModel):
      """Response format for listing collections."""
      collections: List[Collection]
      next_cursor: Optional[str] = None 

• src/mcp_server_replicate/replicate_client.py:550-581 (helper)

  ReplicateClient helper method that performs the actual API call to list collections.

  async def list_collections(self) -> list[dict[str, Any]]:
      """Get list of available model collections.
      
      Returns:
          List of collections with their metadata
          
      Raises:
          Exception: If the API request fails
      """
      if not self.client:
          raise RuntimeError("Client not initialized. Check error property for details.")
  
      try:
          response = await self.http_client.get("/collections")
          response.raise_for_status()
          data = response.json()
          
          return [
              {
                  "name": collection["name"],
                  "slug": collection["slug"],
                  "description": collection.get("description"),
              }
              for collection in data.get("results", [])
          ]
  
      except httpx.HTTPError as err:
          logger.error(f"HTTP error listing collections: {str(err)}")
          raise Exception(f"Failed to list collections: {str(err)}") from err
      except Exception as err:
          logger.error(f"Failed to list collections: {str(err)}")
          raise Exception(f"Failed to list collections: {str(err)}") from err

• src/mcp_server_replicate/models/collection.py:9-15 (schema)

  Pydantic schema for individual Collection used in CollectionList.

  class Collection(BaseModel):
      """A collection of related models on Replicate."""
      name: str = Field(..., description="Name of the collection")
      slug: str = Field(..., description="URL-friendly identifier for the collection")
      description: Optional[str] = Field(None, description="Description of the collection's purpose")
      models: List[Model] = Field(default_factory=list, description="Models in this collection")

• src/mcp_server_replicate/tools/collection_tools.py:11-27 (handler)

  Alternative/duplicate handler in collection_tools.py with explicit name parameter.

  @mcp.tool(
      name="list_collections",
      description="List available model collections on Replicate.",
  )
  async def list_collections() -> CollectionList:
      """List available model collections on Replicate.
      
      Returns:
          CollectionList containing available collections
          
      Raises:
          RuntimeError: If the Replicate client fails to initialize
          Exception: If the API request fails
      """
      async with ReplicateClient() as client:
          result = await client.list_collections()
          return CollectionList(collections=[Collection(**collection) for collection in result])