stackoverflow_mcp/server.py

import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator, List, Optional, Dict, Any
from datetime import datetime

from mcp.server.fastmcp import FastMCP, Context
# Instead of importing Error from mcp.server.fastmcp.tools, we'll define our own Error class
# or we can use standard exceptions for now

from .api import StackExchangeAPI
from .types import (
    SearchByQueryInput,
    SearchByErrorInput,
    GetQuestionInput,
    AdvancedSearchInput,
    SearchResult
)

from .formatter import format_response
from .env import STACK_EXCHANGE_API_KEY

@dataclass
class AppContext:
    api: StackExchangeAPI

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with the Stack Exchange API client.

    Args:
        server (FastMCP): The FastMCP server instance

    Returns:
        AsyncIterator[AppContext]: Context containing the API client
    """
    
    api = StackExchangeAPI(
        api_key=STACK_EXCHANGE_API_KEY,
    )
    try:
        yield AppContext(api=api)
    finally:
        await api.close()
        
mcp = FastMCP(
    "Stack Overflow MCP",
    lifespan=app_lifespan,
    dependencies=["httpx", "python-dotenv"]
)

@mcp.tool()
async def advanced_search(
    query: Optional[str] = None,
    tags: Optional[List[str]] = None,
    excluded_tags: Optional[List[str]] = None,
    min_score: Optional[int] = None,
    title: Optional[str] = None,
    body: Optional[str] = None,
    answers: Optional[int] = None,
    has_accepted_answer: Optional[bool] = None,
    views: Optional[int] = None,
    url: Optional[str] = None,
    user_id: Optional[int] = None,
    is_closed: Optional[bool] = None,
    is_wiki: Optional[bool] = None,
    is_migrated: Optional[bool] = None,
    has_notice: Optional[bool] = None,
    from_date: Optional[datetime] = None,
    to_date: Optional[datetime] = None,
    sort_by: Optional[str] = "votes",
    include_comments: Optional[bool] = False,
    response_format: Optional[str] = "markdown",
    limit: Optional[int] = 5,
    ctx: Context = None
) -> str:
    """Advanced search for Stack Overflow questions with many filter options.
    
    Args:
        query (Optional[str]): Free-form search query
        tags (Optional[List[str]]): List of tags to filter by
        excluded_tags (Optional[List[str]]): List of tags to exclude
        min_score (Optional[int]): Minimum score threshold
        title (Optional[str]): Text that must appear in the title
        body (Optional[str]): Text that must appear in the body
        answers (Optional[int]): Minimum number of answers
        has_accepted_answer (Optional[bool]): Whether questions must have an accepted answer
        views (Optional[int]): Minimum number of views
        url (Optional[str]): URL that must be contained in the post
        user_id (Optional[int]): ID of the user who must own the questions
        is_closed (Optional[bool]): Whether to return only closed or open questions
        is_wiki (Optional[bool]): Whether to return only community wiki questions
        is_migrated (Optional[bool]): Whether to return only migrated questions
        has_notice (Optional[bool]): Whether to return only questions with post notices
        from_date (Optional[datetime]): Earliest creation date
        to_date (Optional[datetime]): Latest creation date
        sort_by (Optional[str]): Field to sort by (activity, creation, votes, relevance)
        include_comments (Optional[bool]): Whether to include comments in results
        response_format (Optional[str]): Format of response ("json" or "markdown")
        limit (Optional[int]): Maximum number of results to return
        ctx (Context): The context is passed automatically by the MCP
        
    Returns:
        str: Formatted search results
    """
    try:
        api = ctx.request_context.lifespan_context.api
        
        ctx.debug(f"Performing advanced search on Stack Overflow")
        if query:
            ctx.debug(f"Query: {query}")
        if body:
            ctx.debug(f"Body: {body}")
        if tags:
            ctx.debug(f"Tags: {', '.join(tags)}")
        if excluded_tags:
            ctx.debug(f"Excluded tags: {', '.join(excluded_tags)}")
        
        results = await api.advanced_search(
            query=query,
            tags=tags,
            excluded_tags=excluded_tags,
            min_score=min_score,
            title=title,
            body=body,
            answers=answers,
            has_accepted_answer=has_accepted_answer,
            views=views,
            url=url,
            user_id=user_id,
            is_closed=is_closed,
            is_wiki=is_wiki,
            is_migrated=is_migrated,
            has_notice=has_notice,
            from_date=from_date,
            to_date=to_date,
            sort_by=sort_by,
            limit=limit,
            include_comments=include_comments
        )
        
        ctx.debug(f"Found {len(results)} results")
        
        return format_response(results, response_format)
    
    except Exception as e:
        ctx.error(f"Error performing advanced search on Stack Overflow: {str(e)}")
        raise RuntimeError(f"Failed to search Stack Overflow: {str(e)}")

@mcp.tool()
async def search_by_query(
    query: str,
    tags: Optional[List[str]] = None,
    excluded_tags: Optional[List[str]] = None,
    min_score: Optional[int] = None,
    title: Optional[str] = None,
    body: Optional[str] = None,
    has_accepted_answer: Optional[bool] = None,
    answers: Optional[int] = None,
    sort_by: Optional[str] = "votes",
    include_comments: Optional[bool] = False,
    response_format: Optional[str] = "markdown",
    limit: Optional[int] = 5,
    ctx: Context = None 
) -> str:
    """Search Stack Overflow for questions matching a query.

    Args:
        query (str): The search query
        tags (Optional[List[str]]): Optional list of tags to filter by (e.g., ["python", "pandas"])
        excluded_tags (Optional[List[str]]): Optional list of tags to exclude
        min_score (Optional[int]): Minimum score threshold for questions
        title (Optional[str]): Text that must appear in the title
        body (Optional[str]): Text that must appear in the body
        has_accepted_answer (Optional[bool]): Whether questions must have an accepted answer
        answers (Optional[int]): Minimum number of answers
        sort_by (Optional[str]): Field to sort by (activity, creation, votes, relevance)
        include_comments (Optional[bool]): Whether to include comments in results
        response_format (Optional[str]): Format of response ("json" or "markdown")
        limit (Optional[int]): Maximum number of results to return
        ctx (Context): The context is passed automatically by the MCP

    Returns:
        str: Formatted search results
    """
    try:
        api = ctx.request_context.lifespan_context.api
        
        ctx.debug(f"Searching Stack Overflow for: {query}")
        
        if tags: 
            ctx.debug(f"Filtering by tags: {', '.join(tags)}")
        if excluded_tags:
            ctx.debug(f"Excluding tags: {', '.join(excluded_tags)}")
        
        results = await api.search_by_query(
            query=query,
            tags=tags,
            excluded_tags=excluded_tags,
            min_score=min_score,
            title=title,
            body=body,
            has_accepted_answer=has_accepted_answer,
            answers=answers,
            sort_by=sort_by,
            limit=limit,
            include_comments=include_comments
        )
        
        ctx.debug(f"Found {len(results)} results")
        
        return format_response(results, response_format)
    
    except Exception as e:
        ctx.error(f"Error searching Stack Overflow: {str(e)}")
        raise RuntimeError(f"Failed to search Stack Overflow: {str(e)}")


@mcp.tool()
async def search_by_error(
    error_message: str,
    language: Optional[str] = None,
    technologies: Optional[List[str]] = None,
    excluded_tags: Optional[List[str]] = None,
    min_score: Optional[int] = None,
    has_accepted_answer: Optional[bool] = None,
    answers: Optional[int] = None,
    include_comments: Optional[bool] = False,
    response_format: Optional[str] = "markdown",
    limit: Optional[int] = 5,
    ctx: Context = None
) -> str:
    """Search Stack Overflow for solutions to an error message

    Args:
        error_message (str): The error message to search for
        language (Optional[str]): Programming language (e.g., "python", "javascript")
        technologies (Optional[List[str]]): Related technologies (e.g., ["react", "django"])
        excluded_tags (Optional[List[str]]): Optional list of tags to exclude
        min_score (Optional[int]): Minimum score threshold for questions
        has_accepted_answer (Optional[bool]): Whether questions must have an accepted answer
        answers (Optional[int]): Minimum number of answers
        include_comments (Optional[bool]): Whether to include comments in results
        response_format (Optional[str]): Format of response ("json" or "markdown")
        limit (Optional[int]): Maximum number of results to return
        ctx (Context): The context is passed automatically by the MCP

    Returns:
        str: Formatted search results
    """
    try:
        api = ctx.request_context.lifespan_context.api
        
        tags = []
        if language:
            tags.append(language.lower())
        if technologies:
            tags.extend([t.lower() for t in technologies])
            
        ctx.debug(f"Searching Stack Overflow for error: {error_message}")
        
        if tags:
            ctx.debug(f"Using tags: {', '.join(tags)}")
        if excluded_tags:
            ctx.debug(f"Excluding tags: {', '.join(excluded_tags)}")
        
        results = await api.search_by_query(
            query=error_message,
            tags=tags if tags else None,
            excluded_tags=excluded_tags,
            min_score=min_score,
            has_accepted_answer=has_accepted_answer,
            answers=answers,
            limit=limit,
            include_comments=include_comments
        )
        ctx.debug(f"Found {len(results)} results")
        
        return format_response(results, response_format)
    except Exception as e: 
        ctx.error(f"Error searching Stack Overflow: {str(e)}")
        raise RuntimeError(f"Failed to search Stack Overflow: {str(e)}")
    
@mcp.tool()
async def get_question(
    question_id: int,
    include_comments: Optional[bool] = True,
    response_format: Optional[str] = "markdown",
    ctx: Context = None
) -> str:
    """Get a specific Stack Overflow question by ID.

    Args:
        question_id (int): The Stack Overflow question ID
        include_comments (Optional[bool]): Whether to include comments in results
        response_format (Optional[str]): Format of response ("json" or "markdown")
        ctx (Context): The context is passed automatically by the MCP

    Returns:
        str: Formatted question details
    """
    try:
        api = ctx.request_context.lifespan_context.api 
        
        ctx.debug(f"Fetching Stack Overflow question: {question_id}")
        
        result = await api.get_question(
            question_id=question_id,
            include_comments=include_comments
        )
        
        return format_response([result], response_format)
    
    except Exception as e:
        ctx.error(f"Error fetching Stack Overflow question: {str(e)}")
        raise RuntimeError(f"Failed to fetch Stack Overflow question: {str(e)}")

@mcp.tool()
async def analyze_stack_trace(
    stack_trace: str,
    language: str,
    excluded_tags: Optional[List[str]] = None,
    min_score: Optional[int] = None,
    has_accepted_answer: Optional[bool] = None,
    answers: Optional[int] = None,
    include_comments: Optional[bool] = True,
    response_format: Optional[str] = "markdown",
    limit: Optional[int] = 3,
    ctx: Context = None
) -> str:
    """Analyze a stack trace and find relevant solutions on Stack Overflow.
    
    Args:
        stack_trace (str): The stack trace to analyze
        language (str): Programming language of the stack trace
        excluded_tags (Optional[List[str]]): Optional list of tags to exclude
        min_score (Optional[int]): Minimum score threshold for questions
        has_accepted_answer (Optional[bool]): Whether questions must have an accepted answer
        answers (Optional[int]): Minimum number of answers
        include_comments (Optional[bool]): Whether to include comments in results
        response_format (Optional[str]): Format of response ("json" or "markdown")
        limit (Optional[int]): Maximum number of results to return
        ctx (Context): The context is passed automatically by the MCP
        
    Returns:
        str: Formatted search results
    """
    try:
        api = ctx.request_context.lifespan_context.api
        
        error_lines = stack_trace.split("\n")
        error_message = error_lines[0]
        
        ctx.debug(f"Analyzing stack trace: {error_message}")
        ctx.debug(f"Language: {language}")
        
        results = await api.search_by_query(
            query=error_message,
            tags=[language.lower()],
            excluded_tags=excluded_tags,
            min_score=min_score,
            has_accepted_answer=has_accepted_answer,
            answers=answers,
            limit=limit,
            include_comments=include_comments
        )
        
        ctx.debug(f"Found {len(results)} results")
        
        return format_response(results, response_format)
    except Exception as e:
        ctx.error(f"Error analyzing stack trace: {str(e)}")
        raise RuntimeError(f"Failed to analyze stack trace: {str(e)}")

if __name__ == "__main__":
    mcp.run()