mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 14:34:17 +02:00
159 lines
4.8 KiB
Python
159 lines
4.8 KiB
Python
from typing import cast
|
|
|
|
import pytest
|
|
from pydantic import AnyUrl
|
|
|
|
from fastmcp.client import Client
|
|
from fastmcp.client.transports import FastMCPTransport
|
|
from fastmcp.server.server import FastMCP
|
|
|
|
|
|
@pytest.fixture
|
|
def fastmcp_server():
|
|
"""Fixture that creates a FastMCP server with tools, resources, and prompts."""
|
|
server = FastMCP("TestServer")
|
|
|
|
# Add a tool
|
|
@server.tool()
|
|
def greet(name: str) -> str:
|
|
"""Greet someone by name."""
|
|
return f"Hello, {name}!"
|
|
|
|
# Add a second tool
|
|
@server.tool()
|
|
def add(a: int, b: int) -> int:
|
|
"""Add two numbers together."""
|
|
return a + b
|
|
|
|
# Add a resource
|
|
@server.resource(uri="data://users")
|
|
async def get_users():
|
|
return ["Alice", "Bob", "Charlie"]
|
|
|
|
# Add a resource template
|
|
@server.resource(uri="data://user/{user_id}")
|
|
async def get_user(user_id: str):
|
|
return {"id": user_id, "name": f"User {user_id}", "active": True}
|
|
|
|
# Add a prompt
|
|
@server.prompt()
|
|
def welcome(name: str) -> str:
|
|
return f"Welcome to FastMCP, {name}!"
|
|
|
|
return server
|
|
|
|
|
|
async def test_list_tools(fastmcp_server):
|
|
"""Test listing tools with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
result = await client.list_tools()
|
|
|
|
# Check that our tools are available
|
|
assert len(result) == 2
|
|
assert set(tool.name for tool in result) == {"greet", "add"}
|
|
|
|
|
|
async def test_call_tool(fastmcp_server):
|
|
"""Test calling a tool with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
result = await client.call_tool("greet", {"name": "World"})
|
|
|
|
# The result content should contain our greeting
|
|
content_str = str(result[0])
|
|
assert "Hello, World!" in content_str
|
|
|
|
|
|
async def test_list_resources(fastmcp_server):
|
|
"""Test listing resources with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
result = await client.list_resources()
|
|
|
|
# Check that our resource is available
|
|
assert len(result) == 1
|
|
assert str(result[0].uri) == "data://users"
|
|
|
|
|
|
async def test_list_prompts(fastmcp_server):
|
|
"""Test listing prompts with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
result = await client.list_prompts()
|
|
|
|
# Check that our prompt is available
|
|
assert len(result) == 1
|
|
assert result[0].name == "welcome"
|
|
|
|
|
|
async def test_get_prompt(fastmcp_server):
|
|
"""Test getting a prompt with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
result = await client.get_prompt("welcome", {"name": "Developer"})
|
|
|
|
# The result should contain our welcome message
|
|
result_str = str(result)
|
|
assert "Welcome to FastMCP, Developer!" in result_str
|
|
|
|
|
|
async def test_read_resource(fastmcp_server):
|
|
"""Test reading a resource with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
# Use the URI from the resource we know exists in our server
|
|
uri = cast(
|
|
AnyUrl, "data://users"
|
|
) # Use cast for type hint only, the URI is valid
|
|
result = await client.read_resource(uri)
|
|
|
|
# The contents should include our user list
|
|
contents_str = str(result[0])
|
|
assert "Alice" in contents_str
|
|
assert "Bob" in contents_str
|
|
assert "Charlie" in contents_str
|
|
|
|
|
|
async def test_client_connection(fastmcp_server):
|
|
"""Test that the client connects and disconnects properly."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
# Before connection
|
|
assert not client.is_connected()
|
|
|
|
# During connection
|
|
async with client:
|
|
assert client.is_connected()
|
|
|
|
# After connection
|
|
assert not client.is_connected()
|
|
|
|
|
|
async def test_resource_template(fastmcp_server):
|
|
"""Test using a resource template with InMemoryClient."""
|
|
client = Client(transport=FastMCPTransport(fastmcp_server))
|
|
|
|
async with client:
|
|
# First, list templates
|
|
result = await client.list_resource_templates()
|
|
|
|
# Check that our template is available
|
|
assert len(result) == 1
|
|
assert "data://user/{user_id}" in result[0].uriTemplate
|
|
|
|
# Now use the template with a specific user_id
|
|
uri = cast(AnyUrl, "data://user/123")
|
|
result = await client.read_resource(uri)
|
|
|
|
# Check the content matches what we expect for the provided user_id
|
|
content_str = str(result[0])
|
|
assert '"id": "123"' in content_str
|
|
assert '"name": "User 123"' in content_str
|
|
assert '"active": true' in content_str
|