sf_extract_dspy_mcp.py

#!/usr/bin/env python3
"""
MCP Server for San Francisco information extraction using DSPy
"""

import dspy
from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Any
import json

# Configure DSPy with your model endpoint
lm = dspy.LM(
    model='openai/Qwen/Qwen3-30B-A3B',
    api_base="http://10.0.0.2:5001/v1",
    api_key='not-needed'
)
dspy.configure(lm=lm)

# DSPy Signatures
class ExtractSFInfo(dspy.Signature):
    """Extract San Francisco-related information from text."""
    
    text: str = dspy.InputField(desc="Text containing information about San Francisco")
    neighborhoods: List[str] = dspy.OutputField(desc="List of San Francisco neighborhoods mentioned")
    landmarks: List[str] = dspy.OutputField(desc="List of San Francisco landmarks or attractions")
    restaurants: List[str] = dspy.OutputField(desc="List of restaurants or food places")
    events: List[str] = dspy.OutputField(desc="List of events or activities")
    transportation: List[str] = dspy.OutputField(desc="Transportation methods or routes mentioned")

class SummarizeSF(dspy.Signature):
    """Summarize San Francisco information."""
    
    extracted_info: str = dspy.InputField(desc="Previously extracted SF information")
    summary: str = dspy.OutputField(desc="Concise summary of San Francisco highlights")
    recommendations: List[str] = dspy.OutputField(desc="Top 3 recommendations for visitors")

class CategorizeSF(dspy.Signature):
    """Categorize San Francisco information by activity type."""
    
    text: str = dspy.InputField()
    outdoor_activities: List[str] = dspy.OutputField(desc="Outdoor activities and attractions")
    cultural_sites: List[str] = dspy.OutputField(desc="Museums, galleries, cultural venues")
    food_scene: List[str] = dspy.OutputField(desc="Restaurants, cafes, food markets")
    neighborhoods: List[str] = dspy.OutputField(desc="Notable neighborhoods and districts")

# DSPy Pipeline
class SanFranciscoExtractor:
    """Multi-stage pipeline for extracting and organizing SF information."""
    
    def __init__(self):
        self.extract_info = dspy.Predict(ExtractSFInfo)
        self.categorize = dspy.Predict(CategorizeSF)
        self.summarize = dspy.ChainOfThought(SummarizeSF)
    
    def __call__(self, text: str):
        # Stage 1: Extract basic SF information
        extracted = self.extract_info(text=text)
        
        # Stage 2: Categorize information
        categorized = self.categorize(text=text)
        
        # Stage 3: Create summary with chain of thought
        info_str = f"""
        Neighborhoods: {extracted.neighborhoods}
        Landmarks: {extracted.landmarks}
        Restaurants: {extracted.restaurants}
        Events: {extracted.events}
        Transportation: {extracted.transportation}
        
        Categories:
        Outdoor: {categorized.outdoor_activities}
        Cultural: {categorized.cultural_sites}
        Food: {categorized.food_scene}
        Districts: {categorized.neighborhoods}
        """
        
        summary = self.summarize(extracted_info=info_str)
        
        return {
            'extracted': extracted,
            'categorized': categorized,
            'summary': summary
        }

# Create MCP server
mcp = FastMCP("San Francisco Info Extractor")

# Initialize the extractor
sf_extractor = SanFranciscoExtractor()

# MCP Tools
@mcp.tool()
def extract_sf_info(text: str) -> str:
    """Extract San Francisco information from text using DSPy pipeline."""
    try:
        result = sf_extractor(text=text)
        
        # Format the result as JSON string for easy parsing
        formatted_result = {
            "neighborhoods": result['extracted'].neighborhoods,
            "landmarks": result['extracted'].landmarks,
            "restaurants": result['extracted'].restaurants,
            "events": result['extracted'].events,
            "transportation": result['extracted'].transportation,
            "outdoor_activities": result['categorized'].outdoor_activities,
            "cultural_sites": result['categorized'].cultural_sites,
            "food_scene": result['categorized'].food_scene,
            "summary": result['summary'].summary,
            "recommendations": result['summary'].recommendations
        }
        
        return json.dumps(formatted_result, indent=2)
    except Exception as e:
        return f"Error processing text: {str(e)}"

@mcp.tool()
def extract_basic_info(text: str) -> str:
    """Extract basic San Francisco information without full pipeline."""
    try:
        extractor = dspy.Predict(ExtractSFInfo)
        result = extractor(text=text)
        
        return json.dumps({
            "neighborhoods": result.neighborhoods,
            "landmarks": result.landmarks,
            "restaurants": result.restaurants,
            "events": result.events,
            "transportation": result.transportation
        }, indent=2)
    except Exception as e:
        return f"Error extracting basic info: {str(e)}"

@mcp.tool()
def categorize_sf_content(text: str) -> str:
    """Categorize San Francisco content by activity type."""
    try:
        categorizer = dspy.Predict(CategorizeSF)
        result = categorizer(text=text)
        
        return json.dumps({
            "outdoor_activities": result.outdoor_activities,
            "cultural_sites": result.cultural_sites,
            "food_scene": result.food_scene,
            "neighborhoods": result.neighborhoods
        }, indent=2)
    except Exception as e:
        return f"Error categorizing content: {str(e)}"

@mcp.tool()
def summarize_sf_info(extracted_info: str) -> str:
    """Generate summary and recommendations from extracted SF information."""
    try:
        summarizer = dspy.ChainOfThought(SummarizeSF)
        result = summarizer(extracted_info=extracted_info)
        
        return json.dumps({
            "summary": result.summary,
            "recommendations": result.recommendations
        }, indent=2)
    except Exception as e:
        return f"Error summarizing: {str(e)}"

# MCP Resources
@mcp.resource("sf://examples")
def get_sf_examples() -> str:
    """Get example San Francisco texts for testing."""
    examples = [
        {
            "title": "Weekend Trip",
            "content": """I spent an amazing weekend in San Francisco exploring different neighborhoods. 
            Started in Chinatown with dim sum at Golden Dragon Restaurant, then walked through 
            North Beach to see Coit Tower. The cable cars were crowded but fun to ride up to 
            Nob Hill. Later visited Alcatraz Island - the audio tour was incredible. 
            Ended the day in Mission District trying tacos at La Taqueria and checking out 
            the street art in Clarion Alley. The Golden Gate Bridge views from Crissy Field 
            were spectacular at sunset."""
        },
        {
            "title": "Food Adventure",
            "content": """San Francisco's food scene is unmatched! Had the best sourdough bread at Boudin Bakery 
            on Fisherman's Wharf, followed by clam chowder in a bread bowl. Spent the afternoon 
            at the Ferry Building Marketplace sampling local produce and artisanal goods. 
            For dinner, tried the famous cioppino at Swan Oyster Depot in Polk Gulch. 
            The Castro District has amazing rainbow crosswalks and historic venues. 
            Don't miss riding the Powell-Mason cable car line - it's like a moving landmark!"""
        },
        {
            "title": "Cultural Tour",
            "content": """Planning a cultural trip to San Francisco. The Museum of Modern Art (SFMOMA) has 
            an incredible contemporary collection. Golden Gate Park houses both the de Young 
            Museum and California Academy of Sciences. Lombard Street is the world's most 
            crooked street and worth the Instagram photo. Union Square is perfect for shopping 
            and people-watching. The Haight-Ashbury neighborhood still has that 1960s hippie vibe. 
            Take the BART train to explore beyond the city center."""
        }
    ]
    
    return json.dumps(examples, indent=2)

@mcp.resource("sf://status")
def get_server_status() -> str:
    """Get the status of the SF extraction server."""
    try:
        # Test DSPy connection
        test_extractor = dspy.Predict(ExtractSFInfo)
        test_result = test_extractor(text="San Francisco is a beautiful city.")
        
        return json.dumps({
            "status": "active",
            "dspy_configured": True,
            "model": lm.model,
            "api_base": lm.api_base,
            "test_extraction": "successful"
        }, indent=2)
    except Exception as e:
        return json.dumps({
            "status": "error",
            "error": str(e)
        }, indent=2)

# MCP Prompts
@mcp.prompt()
def sf_extraction_prompt(user_text: str = "your San Francisco experience") -> str:
    """Create a prompt for San Francisco information extraction."""
    return f"""Please analyze the following text about San Francisco and extract relevant information:

Text to analyze: {user_text}

I need you to identify:
1. Neighborhoods mentioned
2. Landmarks and attractions
3. Restaurants and food places
4. Events and activities
5. Transportation methods

Please provide the extracted information in a structured format."""

@mcp.prompt()
def sf_recommendation_prompt(location_type: str = "restaurant") -> str:
    """Generate a prompt for San Francisco recommendations."""
    return f"""I'm visiting San Francisco and looking for {location_type} recommendations. 
    Please provide detailed suggestions including:
    - Specific names and locations
    - What makes each place special
    - Best times to visit
    - Any insider tips
    
    Focus on authentic local experiences rather than just tourist attractions."""

if __name__ == "__main__":
    # Run the server with SSE transport
    mcp.run(transport="sse")