Coverage for src/pip_project_template/mcp_servers/McpServer02.py: 100.00%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# Timestamp: "2025-08-27 01:43:41 (ywatanabe)" 

4# File: /home/ywatanabe/proj/pip-project-template/src/mcp_servers/McpServer02.py 

5# ---------------------------------------- 

6from __future__ import annotations 

7import os 

8__FILE__ = ( 

9 "./pip-project-template/src/mcp_servers/McpServer02.py" 

10) 

11__DIR__ = os.path.dirname(__FILE__) 

12# ---------------------------------------- 

13 

14"""FastMCP Server 02 implementation with enhanced features.""" 

15 

16from typing import Any, Dict, List 

17 

18from fastmcp import Context, FastMCP 

19 

20from ..core._Calculator import Calculator 

21 

22# Create FastMCP server instance 

23mcp = FastMCP("mcp-server-02") 

24 

25# Initialize calculator 

26calculator = Calculator() 

27 

28 

29@mcp.tool 

30async def calculate_advanced( 

31 a: float, b: float, operation: str = "add", ctx: Context = None 

32) -> Dict[str, Any]: 

33 """Perform mathematical calculations with logging. 

34 

35 Args: 

36 a: First number 

37 b: Second number 

38 operation: Operation to perform (add or multiply) 

39 ctx: FastMCP context for logging 

40 

41 Returns: 

42 Dictionary with calculation results 

43 """ 

44 if ctx: 

45 await ctx.info( 

46 f"Performing {operation} operation: {a} {operation} {b}" 

47 ) 

48 

49 result = calculator.calculate(a, b, operation) 

50 

51 if ctx: 

52 await ctx.info(f"Result calculated: {result}") 

53 

54 return { 

55 "success": True, 

56 "result": result, 

57 "operation": operation, 

58 "inputs": {"a": a, "b": b}, 

59 "server": "McpServer02", 

60 } 

61 

62 

63@mcp.tool 

64def batch_calculate(operations: List[Dict[str, Any]]) -> List[Dict[str, Any]]: 

65 """Perform multiple calculations in batch. 

66 

67 Args: 

68 operations: List of operations, each with 'a', 'b', and 'operation' keys 

69 

70 Returns: 

71 List of calculation results 

72 """ 

73 results = [] 

74 for op in operations: 

75 try: 

76 a = float(op.get("a", 0)) 

77 b = float(op.get("b", 0)) 

78 operation = op.get("operation", "add") 

79 

80 result = calculator.calculate(a, b, operation) 

81 results.append( 

82 { 

83 "success": True, 

84 "result": result, 

85 "operation": operation, 

86 "inputs": {"a": a, "b": b}, 

87 } 

88 ) 

89 except Exception as e: 

90 results.append({"success": False, "error": str(e), "inputs": op}) 

91 

92 return results 

93 

94 

95@mcp.tool 

96def get_server_info() -> Dict[str, Any]: 

97 """Get enhanced information about this MCP server. 

98 

99 Returns: 

100 Dictionary with server information 

101 """ 

102 return { 

103 "server": "McpServer02", 

104 "version": "0.1.0", 

105 "framework": "FastMCP", 

106 "description": "Enhanced mathematical calculation server with batch processing", 

107 "capabilities": [ 

108 "single_calculation", 

109 "batch_calculation", 

110 "async_operations", 

111 "logging", 

112 ], 

113 } 

114 

115 

116@mcp.resource("server://metrics") 

117def server_metrics(): 

118 """Server performance metrics resource.""" 

119 return { 

120 "status": "running", 

121 "server": "McpServer02", 

122 "tools_available": 3, 

123 "capabilities": ["batch_processing", "async_operations"], 

124 "framework": "FastMCP", 

125 } 

126 

127 

128@mcp.resource("calculations://history") 

129def calculation_history(): 

130 """Historical calculation data resource.""" 

131 return { 

132 "recent_operations": [ 

133 {"operation": "add", "count": 42}, 

134 {"operation": "multiply", "count": 28}, 

135 ], 

136 "total_calculations": 70, 

137 "server": "McpServer02", 

138 } 

139 

140 

141def run_server( 

142 transport: str = "stdio", host: str = "localhost", port: int = 8082 

143): 

144 """Run the MCP server with specified transport.""" 

145 if transport == "stdio": 

146 mcp.run(transport="stdio") 

147 elif transport == "http": 

148 mcp.run(transport="http", host=host, port=port, path="/mcp") 

149 elif transport == "sse": 

150 mcp.run(transport="sse", host=host, port=port) 

151 else: 

152 raise ValueError(f"Unsupported transport: {transport}") 

153 

154 

155def main(): 

156 """Run MCP server in STDIO mode.""" 

157 run_server() 

158 

159 

160if __name__ == "__main__": 

161 main() 

162 

163# EOF