PouyaEsmaeili/mcp

from mcp.server.fastmcp import FastMCP

# You can customize important parameters by passing them to FastMCP.
# In this example, all key parameters are set to their default values,
# but you can modify them according to your needs.
# sse_url = http://0.0.0.0:8000/sse
mcp = FastMCP(
    name="MCPServer",
    debug=True,
    host="0.0.0.0",
    port=8000,
    sse_path="/sse",
    message_path="/messages/",
    log_level="DEBUG",
)


@mcp.resource(
    uri="https://quiz.xyz",
    name="GetQuiz",
    description="Provides a link to an online English level assessment quiz.",
)
def get_quiz() -> str:
    return "Link to online quiz: https://quiz.xyz"


@mcp.tool(
    name="FindLevel",
    description="Determines the student's English level based on their quiz score.",
)
def find_level(grade: int) -> str:
    if grade < 50:
        return "Beginner"
    if grade < 75:
        return "Intermediate"
    return "Expert"


@mcp.prompt(
    name="GetPrompt",
    description="Generates a prompt to ask an LLM to teach English based on the student's level.",
)
def get_prompt(name: str, level: str) -> str:
    return f"Teach {name} English based on this level: {level}."

from starlette.applications import Starlette
from starlette.routing import Mount
import uvicorn


if __name__ == "__main__":
    app = Starlette(debug=True, routes=[
        Mount("/", mcp.sse_app()),
    ])
    uvicorn.run(app)

import asyncio

async def main():
    await mcp.run_sse_async()


if __name__ == "__main__":
    asyncio.run(main())

# Pass transport=stdio for stdio server
if __name__ == "__main__":
    mcp.run(transport="sse")

import asyncio
import logging
from typing import Union
from urllib.parse import urlparse

from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream

import mcp.types as types
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.shared.session import RequestResponder

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


async def message_handler(
    message: Union[
        RequestResponder[types.ServerRequest, types.ClientResult],
        types.ServerNotification,
        Exception,
    ],
) -> None:
    if isinstance(message, Exception):
        logger.error("Error: %s", message)
        return
    logger.info("Received message from server: %s", message)


async def run_session(
    read_stream: MemoryObjectReceiveStream,
    write_stream: MemoryObjectSendStream,
) -> None:
    async with ClientSession(
        read_stream,
        write_stream,
        message_handler=message_handler,
    ) as session:
        initialize_response = await session.initialize()
        logger.info(f"Initialized response: {initialize_response}")
        tools = await session.list_tools()
        tools = tools.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")
        response = await session.call_tool("FindLevel", {"grade": 86})
        logger.info(f"Find level response: {response}")


async def main(command_or_url: str) -> None:
    if urlparse(command_or_url).scheme in ("http", "https"):
        async with sse_client(command_or_url) as streams:
            await run_session(*streams)


if __name__ == "__main__":
    sse_url = "http://0.0.0.0:8000/sse"  # change url
    asyncio.run(main(sse_url))

python stdio.py fastmcp-server.py

import asyncio
import sys
import logging
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


class MCPClient:
    def __init__(self) -> None:
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str) -> None:
        if not server_script_path.endswith(".py"):
            raise ValueError("Server script must be a .py file")

        server_params = StdioServerParameters(
            command="python",
            args=[server_script_path],
            env=None,
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport

        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )
        initialize_response = await self.session.initialize()
        logger.info(f"Initialized response: {initialize_response}")

    async def list_tools(self) -> None:
        if not self.session:
            raise RuntimeError("Session is not initialized.")
        response = await self.session.list_tools()
        tools = response.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")

    async def find_level(self, grade: int) -> None:
        response = await self.session.call_tool("FindLevel", {"grade": grade})
        logger.info(f"Find level response: {response}")

    async def cleanup(self) -> None:
        await self.exit_stack.aclose()


async def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.list_tools()
        await client.find_level(86)
    except Exception as exc:
        print("Error:", exc)
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())