Yelp Fusion AI MCP Server

Instructions

Intelligent Yelp business agent designed for agent-to-agent communication.
Handles any natural language request about local businesses through conversational
interaction with Yelp's comprehensive business data and reservation platform.
Returns both natural language responses and structured business data.
Maintains conversation context for multi-turn interactions.

CRITICAL: When recommending businesses, you MUST ALWAYS include the Yelp
URL from the structured data to ensure users can view the business on
Yelp directly.

Capabilities include but are not limited to: business search, detailed
questions, comparisons, itinerary planning, reservation booking
exclusively through the Yelp Reservations platform at participating
restaurants, and any other business-related analysis or recommendations an
intelligent agent could provide with access to Yelp's full dataset.

Use chat_id for follow-up questions and conversational context.

Examples:
- "Find emergency plumbers in Boston"
- "What do people say about the quality of their work?" (follow-up with chat_id)
- "Plan a progressive date in SF's Mission District"
- "What are their hours?" (follow-up with chat_id)
- "Book table for 2 at Mama Nachas tonight at 7pm"
- "Compare auto repair shops from budget to luxury in Sacramento"

Args:
    natural_language_query: Any business-related request in natural language
    search_latitude: Optional latitude coordinate for precise location-based searches
    search_longitude: Optional longitude coordinate for precise location-based searches
    chat_id: Previous response's chat_id for conversational context

Implementation Reference

• src/yelp_agent/main.py:19-78 (handler)

  The core handler function for the 'yelp_agent' tool, decorated with @mcp.tool() for registration. Executes the tool logic by sending the query to Yelp Fusion AI API and formatting the response.

  @mcp.tool()
  async def yelp_agent(
      natural_language_query: str,
      search_latitude: Optional[float] = None,
      search_longitude: Optional[float] = None,
      chat_id: Optional[str] = None,
  ):
      """
      Intelligent Yelp business agent designed for agent-to-agent communication.
      Handles any natural language request about local businesses through conversational
      interaction with Yelp's comprehensive business data and reservation platform.
      Returns both natural language responses and structured business data.
      Maintains conversation context for multi-turn interactions.
  
      CRITICAL: When recommending businesses, you MUST ALWAYS include the Yelp
      URL from the structured data to ensure users can view the business on
      Yelp directly.
  
      Capabilities include but are not limited to: business search, detailed
      questions, comparisons, itinerary planning, reservation booking
      exclusively through the Yelp Reservations platform at participating
      restaurants, and any other business-related analysis or recommendations an
      intelligent agent could provide with access to Yelp's full dataset.
  
      Use chat_id for follow-up questions and conversational context.
  
      Examples:
      - "Find emergency plumbers in Boston"
      - "What do people say about the quality of their work?" (follow-up with chat_id)
      - "Plan a progressive date in SF's Mission District"
      - "What are their hours?" (follow-up with chat_id)
      - "Book table for 2 at Mama Nachas tonight at 7pm"
      - "Compare auto repair shops from budget to luxury in Sacramento"
  
      Args:
          natural_language_query: Any business-related request in natural language
          search_latitude: Optional latitude coordinate for precise location-based searches
          search_longitude: Optional longitude coordinate for precise location-based searches
          chat_id: Previous response's chat_id for conversational context
      """
      logger.info("Interacting with Yelp app with query: %s", natural_language_query)
  
      response = await make_fusion_ai_request(
          natural_language_query,
          user_context=(
              UserContext(
                  latitude=search_latitude,
                  longitude=search_longitude,
              )
              if search_latitude is not None and search_longitude is not None
              else None
          ),
          chat_id=chat_id,
      )
  
      if not response:
          return "Unable to fetch data from Yelp."
  
      return format_fusion_ai_response(response)

• src/yelp_agent/api.py:11-14 (schema)

  TypedDict schema defining UserContext for location-based searches (latitude/longitude). Used in the tool handler.

  class UserContext(TypedDict):
      latitude: float
      longitude: float

• src/yelp_agent/api.py:16-60 (helper)

  Helper function that makes the actual HTTP POST request to Yelp's Fusion AI chat endpoint with authentication and error handling.

  async def make_fusion_ai_request(
      query: str,
      chat_id: Optional[str] = None,
      user_context: Optional[UserContext] = None,
  ):
      yelp_api_key = os.getenv("YELP_API_KEY")
  
      if not yelp_api_key:
          logger.warning(
              "YELP_API_KEY is missing from the environment. Unable to make authorized requests."
          )
          return None
  
      headers = {
          "Authorization": f"Bearer {yelp_api_key}",
          "Content-Type": "application/json",
      }
  
      data: dict[str, Any] = {
          "query": query,
      }
      if chat_id:
          data["chat_id"] = chat_id
      if user_context:
          data["user_context"] = user_context
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.post(
                  url="https://api.yelp.com/ai/chat/v2",
                  json=data,
                  headers=headers,
                  timeout=30.0,
              )
              response.raise_for_status()
              return response.json()
          except httpx.RequestError as e:
              logger.error(f"Request error while making Fusion AI request: {e}")
              return None
          except httpx.HTTPStatusError as e:
              logger.error(f"HTTP status error while making Fusion AI request: {e}")
              return None
          except Exception as e:
              logger.error(f"Unexpected error while making Fusion AI request: {e}")

• src/yelp_agent/formatters.py:6-193 (helper)

  Helper function that parses and formats the API response into structured Markdown output including business details, reviews, hours, and highlights for easy LLM consumption.

  def format_fusion_ai_response(response: dict) -> str:
      """
      Format the response from Fusion AI into a readable string.
      Args:
          response (dict): The response from Fusion AI.
      Returns:
          str: A formatted string containing the response text,
               entities, and Chat ID.
      """
      if not _check_response_format(response):
          logger.error("Invalid response format from Fusion AI.")
          return "Invalid response format from Fusion AI."
      try:
  
          data = response
          # Get the response text for introduction
          intro_text = (
              data.get("response", {})
              .get("text", "Business information")
              .replace("[[HIGHLIGHT]]", "**")
              .replace("[[ENDHIGHLIGHT]]", "**")
          )
  
          # Chat ID
          chat_id = data.get("chat_id", "Unknown Chat ID")
  
          # Initialize the formatted output
          formatted_output = (
              "# Formatted Business Data for LLM Processing\n\n"
              "## Introduction\n"
              f"{intro_text}\n \n"
              "## Chat ID\n"
              f"{chat_id}\n\n"
          )
  
          # Check if entities and businesses exist
          businesses = []
          for entity in data.get("entities", []):
              if "businesses" in entity:
                  businesses = entity["businesses"]
                  break
  
          logger.debug("Found %d businesses for the query", len(businesses))
  
          # Process each Business
          for index, business in enumerate(businesses):
              name = business.get("name", "Unknown")
              formatted_output += f"\n## Business {index + 1}: {name}\n"
  
              # Rating and reviews
              rating = business.get("rating")
              review_count = business.get("review_count")
              price = business.get("price", "")
              if price:
                  formatted_output += f"- **Price**: {price}\n"
              else:
                  formatted_output += "- **Price**: Not available\n"
              if rating:
                  review_info = f"{rating}/5"
                  if review_count:
                      review_info += f" ({review_count} reviews)"
                  formatted_output += f"- **Rating**: {review_info}\n"
  
              # Categories
              categories = business.get("categories", [])
              if categories:
                  cat_titles = [
                      cat.get("title") for cat in categories if cat.get("title")
                  ]
                  formatted_output += f"- **Type**: {', '.join(cat_titles)}\n"
  
              # Location
              location = business.get("location", {})
              if location:
                  address_parts = location.get("formatted_address", "").split("\n")
                  address = ", ".join(filter(None, address_parts))
                  formatted_output += f"- **Location**: {address}\n"
  
              # Coordinates
              coordinates = business.get("coordinates", {})
              if coordinates:
                  lat = coordinates.get("latitude", "")
                  lon = coordinates.get("longitude", "")
                  if lat and lon:
                      formatted_output += f"- **Coordinates**: {lat}, {lon}\n"
                  else:
                      formatted_output += "- **Coordinates**: Not available\n"
              else:
                  formatted_output += "- **Coordinates**: Not available\n"
  
              # URL
              url = business.get("url", "")
              if url:
                  formatted_output += f"- **URL**: [View on Yelp]({url})\n"
  
              # Phone
              phone = business.get("phone", "")
              if phone:
                  formatted_output += f"- **Phone**: {phone}\n"
  
              # Website
              website = business.get("attributes", {}).get("BusinessUrl")
              if website:
                  formatted_output += f"- **Website**: {website}\n"
  
              # Services
              services = []
  
              if attributes := business.get("attributes", {}):
                  formatted_attributes = _format_business_attributes(attributes)
                  if formatted_attributes:
                      services.append(formatted_attributes)
  
              if services:
                  formatted_output += (
                      "- **Services and Amenities**: \n  - "
                      + "\n  - ".join(services)
                      + "\n"
                  )
  
              # Contextual Info
              contextual_info = business.get("contextual_info", {})
  
              # Business hours
              day_wise_business_hours = contextual_info.get("business_hours", [])
              if day_wise_business_hours and len(day_wise_business_hours) > 0:
                  formatted_output += "- **Hours**:\n"
                  for day in day_wise_business_hours:
                      day_name = day.get("day_of_week", "Unknown")
                      business_hours = day.get("business_hours", [])
                      if business_hours:
                          open_time = business_hours[0].get("open_time", "")
                          close_time = business_hours[0].get("close_time", "")
                          try:
                              open_dt = datetime.datetime.fromisoformat(open_time)
                              close_dt = datetime.datetime.fromisoformat(close_time)
                              open_str = open_dt.strftime("%I:%M %p")
                              close_str = close_dt.strftime("%I:%M %p")
                              formatted_output += (
                                  f"  - {day_name}: {open_str} - {close_str}\n"
                              )
                          except ValueError:
                              formatted_output += f"  - {day_name}: Available\n"
  
              # Overall Review Snippet
              overall_review_snippet = contextual_info.get("review_snippet")
              if overall_review_snippet:
                  review_text = overall_review_snippet.replace(
                      "[[HIGHLIGHT]]", "**"
                  ).replace("[[ENDHIGHLIGHT]]", "**")
                  formatted_output += f"- **Review Highlight**: {review_text}\n"
  
              # Individual Review Snippets
              review_snippets = contextual_info.get("review_snippets", [])
              if review_snippets:
                  formatted_output += "- **Customer Reviews**:\n"
                  for snippet in review_snippets:
                      rating = snippet.get("rating")
                      comment = snippet.get("comment", "No comment.")
                      # Replace highlight markers with markdown bold
                      comment = comment.replace("[[HIGHLIGHT]]", "**").replace(
                          "[[ENDHIGHLIGHT]]", "**"
                      )
                      if rating:
                          formatted_output += f"  - Rating: {rating}/5\n"
                          formatted_output += f"    {comment}\n"
                      else:
                          formatted_output += f"  - {comment}\n"
  
              # Photos
              photos = contextual_info.get("photos", [])
              if photos:
                  formatted_output += "- **Photos**:\n"
                  for photo in photos:
                      photo_url = photo.get("original_url")
                      if photo_url:
                          formatted_output += f"  - {photo_url}\n"
  
              # Description from summaries
              summaries = business.get("summaries", {})
              long_summary = summaries.get("long", "")
              short_summary = summaries.get("short", "")
              description = long_summary or short_summary
              if description:
                  formatted_output += f"- **Description**: {description}\n"
  
          logger.debug("Formatted output for LLM:\n%s", formatted_output)
          return formatted_output

• src/yelp_agent/main.py:19-78 (registration)

  The @mcp.tool() decorator registers the yelp_agent function as an MCP tool.

  @mcp.tool()
  async def yelp_agent(
      natural_language_query: str,
      search_latitude: Optional[float] = None,
      search_longitude: Optional[float] = None,
      chat_id: Optional[str] = None,
  ):
      """
      Intelligent Yelp business agent designed for agent-to-agent communication.
      Handles any natural language request about local businesses through conversational
      interaction with Yelp's comprehensive business data and reservation platform.
      Returns both natural language responses and structured business data.
      Maintains conversation context for multi-turn interactions.
  
      CRITICAL: When recommending businesses, you MUST ALWAYS include the Yelp
      URL from the structured data to ensure users can view the business on
      Yelp directly.
  
      Capabilities include but are not limited to: business search, detailed
      questions, comparisons, itinerary planning, reservation booking
      exclusively through the Yelp Reservations platform at participating
      restaurants, and any other business-related analysis or recommendations an
      intelligent agent could provide with access to Yelp's full dataset.
  
      Use chat_id for follow-up questions and conversational context.
  
      Examples:
      - "Find emergency plumbers in Boston"
      - "What do people say about the quality of their work?" (follow-up with chat_id)
      - "Plan a progressive date in SF's Mission District"
      - "What are their hours?" (follow-up with chat_id)
      - "Book table for 2 at Mama Nachas tonight at 7pm"
      - "Compare auto repair shops from budget to luxury in Sacramento"
  
      Args:
          natural_language_query: Any business-related request in natural language
          search_latitude: Optional latitude coordinate for precise location-based searches
          search_longitude: Optional longitude coordinate for precise location-based searches
          chat_id: Previous response's chat_id for conversational context
      """
      logger.info("Interacting with Yelp app with query: %s", natural_language_query)
  
      response = await make_fusion_ai_request(
          natural_language_query,
          user_context=(
              UserContext(
                  latitude=search_latitude,
                  longitude=search_longitude,
              )
              if search_latitude is not None and search_longitude is not None
              else None
          ),
          chat_id=chat_id,
      )
  
      if not response:
          return "Unable to fetch data from Yelp."
  
      return format_fusion_ai_response(response)