Yelp Fusion AI MCP Server

Instructions

Implementation Reference

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

  The core handler function for the 'yelp_agent' tool. Registered via @mcp.tool() decorator. Processes natural language queries about businesses using Yelp's Fusion AI API, handles location context and chat history, and returns formatted responses.
  @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 for UserContext used to pass location parameters to the Fusion AI request.
  class UserContext(TypedDict): latitude: float longitude: float
• src/yelp_agent/api.py:16-60 (helper)

  Helper function that makes the HTTP POST request to Yelp's Fusion AI /ai/chat/v2 endpoint with the query, context, and chat_id.
  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-197 (helper)

  Helper function that formats the raw JSON response from Fusion AI into a structured markdown string suitable for LLM processing, including business details, reviews, hours, and critical Yelp URLs.
  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 except (KeyError, TypeError, ValueError) as e: logger.error(e) return "Unable to fetch data from Yelp. Invalid response format."