Model Context Protocol (MCP) in Practice

from mcp.server.fastmcp import FastMCP, Context

from newsapi.connector import NewsAPIConnector
from newsapi.models import NewsResponse

mcp = FastMCP("News Server")
news_api_connector = NewsAPIConnector()

@mcp.tool()
async def fetch_news(
    query: str,
    from_date: str | None,
    to_date: str | None,
    language: str = "en",
    sort_by: str = "publishedAt",
    page_size: int = 10,
    page: int = 1,
    ctx: Context = None,
) -> NewsResponse:
    """
    Retrieves news articles.

    Args:
        query: Keywords or phrases to search for
        from_date: Optional start date (YYYY-MM-DD)
        to_date: Optional end date (YYYY-MM-DD)
        language: Language code (default: en)
        sort_by: Sort order (default: publishedAt, popularity, relevancy)
        page_size: Number of results per page (default: 10, max: 100)
        page: Page number for pagination (default: 1)
        ctx: MCP Context

    Returns:
        News Articles
    """
    if ctx:
        ctx.info(f"Searching news for: {query}")

    params = {
        k: v
        for k, v in {
            "q": query,
            "from": from_date,
            "to": to_date,
            "language": language,
            "sortBy": sort_by,
            "pageSize": min(page_size, 100),
            "page": page,
        }.items()
        if v is not None
    }

    success, result = await news_api_connector.search_everything(**params)
    if not success:
        return error_response("Failed to fetch news articles.")
    return result

@mcp.tool()
async def fetch_headlines(
    category: str,
    query: str,
    country: str = "us",
    page_size: int = 10,
    page: int = 1,
    ctx: Context = None,
) -> NewsResponse:
  
  # See fetch_headlines @tool implementation in ./src/server.py

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

class NewsAPIConnector:
    """
    Handles a connection with the NewsAPI and retrieves news articles.
    """

    def __init__(self):
        self.api_key = os.getenv("NEWSAPI_KEY")
        self.base_url = os.getenv("NEWSAPI_URL")

    async def get_client(self) -> httpx.AsyncClient:
        return httpx.AsyncClient(timeout=30.0, headers={"X-Api-Key": self.api_key})

    async def search_everything(
        self, **kwargs
    ) -> Tuple[bool, Union[str, NewsResponse]]:
        """
        Retrieves News Articles from the NewsAPI "/everything" endpoint.

        Args:
            **kwargs: Requests parameters (see https://newsapi.org/docs/endpoints/everything)

        Returns:
            A tuple containing (success, result) where:
            - success: A boolean indicating if the request was successful
            - result: Either an error message (string) or the validated NewsResponse model
        """
        async with await self.get_client() as client:
            params = kwargs
            try:
                response = await client.get(f"{self.base_url}everything", params=params)
                response.raise_for_status()
                data = response.json()
                result = NewsResponse.model_validate(data)

                return True, result

            except httpx.RequestError as e:
                return False, f"Request error: {str(e)}"

            except Exception as e:
                return False, f"Unexpected error: {str(e)}"

async def get_top_headlines(
        self, **kwargs
    ) -> Tuple[bool, Union[str, NewsResponse]]:
        
        # See get_top_headlines implementation in ./newsapi/connector.py

from datetime import datetime
from pydantic import BaseModel, ConfigDict, HttpUrl
from pydantic.alias_generators import to_camel


class ArticleSource(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    id: str | None
    name: str | None


class Article(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    source: ArticleSource
    author: str | None
    title: str | None
    description: str | None
    url: HttpUrl | None
    url_to_image: HttpUrl | None
    published_at: datetime | None
    content: str | None


class NewsResponse(BaseModel):
    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)

    status: str
    totalResults: int
    articles: list[Article]

@mcp.resource(
    "memo://topics/",
    mime_type="application/json"
)
def list_docs():
    return

@mcp.resource(
    "memo://topics/{doc_id}",
    mime_type="text/plain"
)
def fetch_doc(doc_id: str):
    return

@mcp.prompt(
    name:"topic-news-analysis",
    description:"analyzes the news development and sentiment evolution in a certain topic during a certain period"
)
def analyze_topic():
    return ""

from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
 
# Commands for running/connecting to MCP Server
server_params = StdioServerParameters(
    command="python",    # Executable
    args=["server.py"],  # Optional command line arguments
    env=None,            # Optional environment variables
)
 
async with stdio_client(server_params) as (read, write):
    async with ClientSession(
        read, write, sampling_callback=handle_sampling_message
    ) as session:
        await session.initialize()
 
        # List available prompts
        prompts = await session.list_prompts()
 
        # Get a prompt
        prompt = await session.get_prompt(
            "example-prompt", arguments={"arg1": "value"}
        )
 
        # List available resources
        resources = await session.list_resources()
 
        # List available tools
        tools = await session.list_tools()
 
        # Read a resource
        content, mime_type = await session.read_resource("file://some/path")
 
        # Call a tool
        result = await session.call_tool("fetch_news", arguments={"q": "news topic or query"})

{
    "mcpServers": {
        "news": {
            "command": "uv",
            "args": [
                "--directory",
                "C:\\ABSOLUTE\\PATH\\TO\\PARENT\\FOLDER\\news",
                "run",
                "server.py"
            ]
        }
    }
}

@misc{carpintero-mcp-in-practice,
  title = {Model Context Protocol (MCP) in Practice},
  author = {Diego Carpintero},
  month = {apr},
  year = {2025},
  date = {2025-04-20},
  publisher = {https://tech.dcarpintero.com/},
  howpublished = {\url{https://tech.diegocarpintero.com/blog/model-context-protocol-in-practice/}},
  keywords = {large-language-models, agents, model-context-protocol, ai-tools, function-calling},
}