Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add end-to-end tests for server-client communication #294

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 65 additions & 0 deletions tests/test_e2e.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
from collections.abc import AsyncGenerator

import pytest

from mcp import ClientSession, StdioServerParameters, stdio_client
from mcp.server import FastMCP

params = StdioServerParameters(command="uv", args=["run", __file__])


def server() -> FastMCP:
mcp = FastMCP("Echo")

@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""Echo a message as a resource"""
return f"Resource echo: {message}"

@mcp.tool()
def echo_tool(message: str) -> str:
"""Echo a message as a tool"""
return f"Tool echo: {message}"

@mcp.prompt()
def echo_prompt(message: str) -> str:
"""Create an echo prompt"""
return f"Please process this message: {message}"

return mcp


@pytest.fixture
async def mcp_client_session() -> AsyncGenerator[ClientSession, None]:
async with stdio_client(params) as streams:
async with ClientSession(streams[0], streams[1]) as session:
await session.initialize()
yield session


@pytest.mark.anyio
async def test_list_resource_templates(mcp_client_session: ClientSession) -> None:
res = await mcp_client_session.list_resource_templates()
templates = set(template.name for template in res.resourceTemplates)

assert "echo_resource" in templates


@pytest.mark.anyio
async def test_list_tools(mcp_client_session: ClientSession) -> None:
res = await mcp_client_session.list_tools()
tools = set(tool.name for tool in res.tools)

assert "echo_tool" in tools


@pytest.mark.anyio
async def test_list_prompts(mcp_client_session: ClientSession) -> None:
res = await mcp_client_session.list_prompts()
prompts = set(prompt.name for prompt in res.prompts)

assert "echo_prompt" in prompts


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