Improve code quality and fix tests

- Improve readability in func_metadata.py by using helper functions and clear if-statements
- Restore complex_inputs.py example to original state from main branch
- Fix test_88_random_error.py by adding list_tools handler
- Refactor test_lowlevel_tool_output.py to test server behavior directly without clients
- Fix README formatting for structured output examples
- Add type assertions to fix pyright errors in tests

This completes the code improvements requested in the PR review.

Author: bhosmer-ant

README.md

@@ -255,6 +255,7 @@ async def fetch_weather(city: str) -> str:
 Tools can return structured data with automatic validation using the `structured_output=True` parameter. This ensures your tools return well-typed, validated data that clients can easily process:
 
 ```python
+from mcp.server.fastmcp import FastMCP
 from pydantic import BaseModel, Field
 from typing import TypedDict
 
@@ -273,10 +274,7 @@ class WeatherData(BaseModel):
 def get_weather(city: str) -> WeatherData:
     """Get structured weather data"""
     return WeatherData(
-        temperature=22.5,
-        humidity=65.0,
-        condition="partly cloudy",
-        wind_speed=12.3
+        temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3
     )
 
 
@@ -290,22 +288,14 @@ class LocationInfo(TypedDict):
 @mcp.tool(structured_output=True)
 def get_location(address: str) -> LocationInfo:
     """Get location coordinates"""
-    return LocationInfo(
-        latitude=51.5074,
-        longitude=-0.1278,
-        name="London, UK"
-    )
+    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="London, UK")
 
 
 # Using dict[str, Any] for flexible schemas
 @mcp.tool(structured_output=True)
 def get_statistics(data_type: str) -> dict[str, float]:
     """Get various statistics"""
-    return {
-        "mean": 42.5,
-        "median": 40.0,
-        "std_dev": 5.2
-    }
+    return {"mean": 42.5, "median": 40.0, "std_dev": 5.2}
 
 
 # Lists and other types are wrapped automatically

examples/fastmcp/complex_inputs.py

@@ -1,11 +1,9 @@
 """
 FastMCP Complex inputs Example
 
-Demonstrates validation via pydantic with complex models,
-and structured output for returning validated data.
+Demonstrates validation via pydantic with complex models.
 """
 
-from datetime import datetime
 from typing import Annotated
 
 from pydantic import BaseModel, Field
@@ -18,12 +16,8 @@
 class ShrimpTank(BaseModel):
     class Shrimp(BaseModel):
         name: Annotated[str, Field(max_length=10)]
-        color: str = "red"
-        age_days: int = 0
 
     shrimp: list[Shrimp]
-    temperature: float = Field(default=24.0, ge=20.0, le=28.0)
-    ph_level: float = Field(default=7.0, ge=6.5, le=8.0)
 
 
 @mcp.tool()
@@ -34,140 +28,3 @@ def name_shrimp(
 ) -> list[str]:
     """List all shrimp names in the tank"""
     return [shrimp.name for shrimp in tank.shrimp] + extra_names
-
-
-# Structured output example - returns a validated tank analysis
-class TankAnalysis(BaseModel):
-    """Analysis of shrimp tank conditions"""
-    total_shrimp: int
-    temperature_status: str  # "optimal", "too_cold", "too_hot"
-    ph_status: str  # "optimal", "too_acidic", "too_basic"
-    shrimp_by_color: dict[str, int]
-    oldest_shrimp: str | None
-    recommendations: list[str]
-
-
-@mcp.tool(structured_output=True)
-def analyze_tank(tank: ShrimpTank) -> TankAnalysis:
-    """Analyze tank conditions and provide recommendations"""
-    # Temperature analysis
-    if tank.temperature < 22:
-        temp_status = "too_cold"
-    elif tank.temperature > 26:
-        temp_status = "too_hot"
-    else:
-        temp_status = "optimal"
-    
-    # pH analysis
-    if tank.ph_level < 6.8:
-        ph_status = "too_acidic"
-    elif tank.ph_level > 7.5:
-        ph_status = "too_basic"
-    else:
-        ph_status = "optimal"
-    
-    # Count shrimp by color
-    color_counts: dict[str, int] = {}
-    for shrimp in tank.shrimp:
-        color_counts[shrimp.color] = color_counts.get(shrimp.color, 0) + 1
-    
-    # Find oldest shrimp
-    oldest = None
-    if tank.shrimp:
-        oldest_shrimp_obj = max(tank.shrimp, key=lambda s: s.age_days)
-        oldest = oldest_shrimp_obj.name
-    
-    # Generate recommendations
-    recommendations = []
-    if temp_status == "too_cold":
-        recommendations.append("Increase water temperature to 22-26°C")
-    elif temp_status == "too_hot":
-        recommendations.append("Decrease water temperature to 22-26°C")
-    
-    if ph_status == "too_acidic":
-        recommendations.append("Add crushed coral or baking soda to raise pH")
-    elif ph_status == "too_basic":
-        recommendations.append("Add Indian almond leaves or driftwood to lower pH")
-    
-    if len(tank.shrimp) > 20:
-        recommendations.append("Consider dividing colony to prevent overcrowding")
-    
-    if not recommendations:
-        recommendations.append("Tank conditions are optimal!")
-    
-    return TankAnalysis(
-        total_shrimp=len(tank.shrimp),
-        temperature_status=temp_status,
-        ph_status=ph_status,
-        shrimp_by_color=color_counts,
-        oldest_shrimp=oldest,
-        recommendations=recommendations
-    )
-
-
-# Another structured output example - breeding recommendations
-@mcp.tool(structured_output=True)
-def get_breeding_pairs(tank: ShrimpTank) -> dict[str, list[str]]:
-    """Suggest breeding pairs by color
-    
-    Returns a dictionary mapping colors to lists of shrimp names
-    that could be bred together.
-    """
-    pairs_by_color: dict[str, list[str]] = {}
-    
-    for shrimp in tank.shrimp:
-        if shrimp.age_days >= 60:  # Mature enough to breed
-            if shrimp.color not in pairs_by_color:
-                pairs_by_color[shrimp.color] = []
-            pairs_by_color[shrimp.color].append(shrimp.name)
-    
-    # Only return colors with at least 2 shrimp
-    return {
-        color: names 
-        for color, names in pairs_by_color.items() 
-        if len(names) >= 2
-    }
-
-
-if __name__ == "__main__":
-    # For testing the tools
-    import asyncio
-    
-    async def test():
-        # Create a test tank
-        tank = ShrimpTank(
-            shrimp=[
-                ShrimpTank.Shrimp(name="Rex", color="red", age_days=90),
-                ShrimpTank.Shrimp(name="Blue", color="blue", age_days=45),
-                ShrimpTank.Shrimp(name="Crimson", color="red", age_days=120),
-                ShrimpTank.Shrimp(name="Azure", color="blue", age_days=80),
-                ShrimpTank.Shrimp(name="Jade", color="green", age_days=30),
-                ShrimpTank.Shrimp(name="Ruby", color="red", age_days=75),
-            ],
-            temperature=23.5,
-            ph_level=7.2
-        )
-        
-        # Test name_shrimp (non-structured output)
-        names = name_shrimp(tank, ["Bonus1", "Bonus2"])
-        print("Shrimp names:", names)
-        
-        # Test analyze_tank (structured output)
-        print("\nTank Analysis:")
-        analysis = analyze_tank(tank)
-        print(analysis.model_dump_json(indent=2))
-        
-        # Test get_breeding_pairs (structured output returning dict)
-        print("\nBreeding Pairs:")
-        pairs = get_breeding_pairs(tank)
-        print(f"Mature shrimp by color: {pairs}")
-        
-        # Show the tools that would be exposed
-        print("\nAvailable tools:")
-        tools = await mcp.list_tools()
-        for tool in tools:
-            print(f"- {tool.name}: {tool.description}")
-            if tool.outputSchema:
-                print(f"  Output schema: {tool.outputSchema.get('title', tool.outputSchema.get('type', 'structured'))}")
-    
-    asyncio.run(test())

examples/fastmcp/weather_structured.py

@@ -20,6 +20,7 @@
 # Example 1: Using a Pydantic model for structured output
 class WeatherData(BaseModel):
     """Structured weather data response"""
+
     temperature: float = Field(description="Temperature in Celsius")
     humidity: float = Field(description="Humidity percentage (0-100)")
     condition: str = Field(description="Weather condition (sunny, cloudy, rainy, etc.)")
@@ -32,18 +33,13 @@ class WeatherData(BaseModel):
 def get_weather(city: str) -> WeatherData:
     """Get current weather for a city with full structured data"""
     # In a real implementation, this would fetch from a weather API
-    return WeatherData(
-        temperature=22.5,
-        humidity=65.0,
-        condition="partly cloudy",
-        wind_speed=12.3,
-        location=city
-    )
+    return WeatherData(temperature=22.5, humidity=65.0, condition="partly cloudy", wind_speed=12.3, location=city)
 
 
 # Example 2: Using TypedDict for a simpler structure
 class WeatherSummary(TypedDict):
     """Simple weather summary"""
+
     city: str
     temp_c: float
     description: str
@@ -52,27 +48,19 @@ class WeatherSummary(TypedDict):
 @mcp.tool(structured_output=True)
 def get_weather_summary(city: str) -> WeatherSummary:
     """Get a brief weather summary for a city"""
-    return WeatherSummary(
-        city=city,
-        temp_c=22.5,
-        description="Partly cloudy with light breeze"
-    )
+    return WeatherSummary(city=city, temp_c=22.5, description="Partly cloudy with light breeze")
 
 
 # Example 3: Using dict[str, Any] for flexible schemas
 @mcp.tool(structured_output=True)
 def get_weather_metrics(cities: list[str]) -> dict[str, dict[str, float]]:
     """Get weather metrics for multiple cities
-    
+
     Returns a dictionary mapping city names to their metrics
     """
     # Returns nested dictionaries with weather metrics
     return {
-        city: {
-            "temperature": 20.0 + i * 2,
-            "humidity": 60.0 + i * 5,
-            "pressure": 1013.0 + i * 0.5
-        }
+        city: {"temperature": 20.0 + i * 2, "humidity": 60.0 + i * 5, "pressure": 1013.0 + i * 0.5}
         for i, city in enumerate(cities)
     }
 
@@ -81,6 +69,7 @@ def get_weather_metrics(cities: list[str]) -> dict[str, dict[str, float]]:
 @dataclass
 class WeatherAlert:
     """Weather alert information"""
+
     severity: str  # "low", "medium", "high"
     title: str
     description: str
@@ -99,15 +88,15 @@ def get_weather_alerts(region: str) -> list[WeatherAlert]:
                 title="Heat Wave Warning",
                 description="Temperatures expected to exceed 40°C",
                 affected_areas=["Los Angeles", "San Diego", "Riverside"],
-                valid_until=datetime(2024, 7, 15, 18, 0)
+                valid_until=datetime(2024, 7, 15, 18, 0),
             ),
             WeatherAlert(
                 severity="medium",
                 title="Air Quality Advisory",
                 description="Poor air quality due to wildfire smoke",
                 affected_areas=["San Francisco Bay Area"],
-                valid_until=datetime(2024, 7, 14, 12, 0)
-            )
+                valid_until=datetime(2024, 7, 14, 12, 0),
+            ),
         ]
     return []
 
@@ -116,26 +105,28 @@ def get_weather_alerts(region: str) -> list[WeatherAlert]:
 @mcp.tool(structured_output=True)
 def get_temperature(city: str, unit: str = "celsius") -> float:
     """Get just the temperature for a city
-    
+
     When returning primitives with structured_output=True,
     the result is wrapped in {"result": value}
     """
     base_temp = 22.5
     if unit.lower() == "fahrenheit":
-        return base_temp * 9/5 + 32
+        return base_temp * 9 / 5 + 32
     return base_temp
 
 
 # Example 6: Weather statistics with nested models
 class DailyStats(BaseModel):
     """Statistics for a single day"""
+
     high: float
     low: float
     mean: float
-    
-    
+
+
 class WeatherStats(BaseModel):
     """Weather statistics over a period"""
+
     location: str
     period_days: int
     temperature: DailyStats
@@ -151,32 +142,32 @@ def get_weather_stats(city: str, days: int = 7) -> WeatherStats:
         period_days=days,
         temperature=DailyStats(high=28.5, low=15.2, mean=21.8),
         humidity=DailyStats(high=85.0, low=45.0, mean=65.0),
-        precipitation_mm=12.4
+        precipitation_mm=12.4,
     )
 
 
 if __name__ == "__main__":
     # For testing individual tools
     import asyncio
-    
+
     async def test():
         # Test the tools
         weather = get_weather("London")
         print(f"Weather in London: {weather.model_dump_json(indent=2)}")
-        
+
         summary = get_weather_summary("Paris")
         print(f"\nWeather summary for Paris: {summary}")
-        
+
         metrics = get_weather_metrics(["Tokyo", "Sydney", "Mumbai"])
         print(f"\nWeather metrics: {metrics}")
-        
+
         alerts = get_weather_alerts("California")
         print(f"\nWeather alerts: {len(alerts)} alerts found")
-        
+
         temp = get_temperature("Berlin", "fahrenheit")
         print(f"\nTemperature in Berlin: {temp}°F")
-        
+
         stats = get_weather_stats("Seattle", 30)
         print(f"\nWeather stats for Seattle: {stats.model_dump_json(indent=2)}")
-    
-    asyncio.run(test())
+
+    asyncio.run(test())

src/mcp/client/session.py

@@ -396,6 +396,7 @@ async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult:
         )
 
         # Cache tool output schemas for future validation
+        # Note: don't clear the cache, as we may be using a cursor
         for tool in result.tools:
             self._tool_output_schemas[tool.name] = tool.outputSchema