yelp_agent
Handle natural language queries about local businesses through conversational interaction with Yelp's business data, providing search, recommendations, comparisons, and reservation booking.
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 contextInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| natural_language_query | Yes | ||
| search_latitude | No | ||
| search_longitude | No | ||
| chat_id | No |
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)