Coverage for src / mcp_server_langgraph / tools / calculator_tools.py: 75%
100 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""
2Calculator tools for mathematical operations
4Provides basic arithmetic and evaluation tools for the agent.
5"""
7import ast
8import operator
9from typing import Any
11from langchain_core.tools import tool
12from pydantic import Field
15def _safe_log(level: str, message: str, **kwargs: Any) -> None:
16 """Safely log message, handling cases where observability isn't initialized"""
17 try:
18 from mcp_server_langgraph.observability.telemetry import logger
20 getattr(logger, level)(message, **kwargs)
21 except (ImportError, RuntimeError):
22 # Observability not available - silently skip
23 pass
26def _safe_metric(metric_name: str, value: float, attributes: dict[str, Any]) -> None:
27 """Safely record metric, handling cases where observability isn't initialized"""
28 try:
29 from mcp_server_langgraph.observability.telemetry import metrics
31 getattr(metrics, metric_name).add(value, attributes)
32 except (ImportError, RuntimeError):
33 # Observability not available - silently skip
34 pass
37# Safe operators for calculator evaluation
38SAFE_OPERATORS = {
39 ast.Add: operator.add,
40 ast.Sub: operator.sub,
41 ast.Mult: operator.mul,
42 ast.Div: operator.truediv,
43 ast.Pow: operator.pow,
44 ast.USub: operator.neg,
45 ast.UAdd: operator.pos,
46 ast.Mod: operator.mod,
47}
50def _safe_eval(expression: str) -> float:
51 """
52 Safely evaluate a mathematical expression.
54 Args:
55 expression: Mathematical expression (e.g., "2 + 2", "10 * 3")
57 Returns:
58 Result of evaluation
60 Raises:
61 ValueError: If expression contains unsafe operations
62 SyntaxError: If expression is malformed
63 """
65 def _eval(node: Any) -> float:
66 if isinstance(node, ast.Constant): # Number (Python 3.8+)
67 if isinstance(node.value, (int, float)): 67 ↛ 70line 67 didn't jump to line 70 because the condition on line 67 was always true
68 return float(node.value)
69 else:
70 msg = f"Invalid constant type: {type(node.value).__name__}"
71 raise ValueError(msg)
72 elif isinstance(node, ast.BinOp): # Binary operation
73 op_func = SAFE_OPERATORS.get(type(node.op))
74 if op_func is None: 74 ↛ 75line 74 didn't jump to line 75 because the condition on line 74 was never true
75 msg = f"Unsafe operator: {type(node.op).__name__}"
76 raise ValueError(msg)
77 return float(op_func(_eval(node.left), _eval(node.right))) # type: ignore[operator]
78 elif isinstance(node, ast.UnaryOp): # Unary operation 78 ↛ 85line 78 didn't jump to line 85 because the condition on line 78 was always true
79 op_func = SAFE_OPERATORS.get(type(node.op))
80 if op_func is None: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 msg = f"Unsafe operator: {type(node.op).__name__}"
82 raise ValueError(msg)
83 return float(op_func(_eval(node.operand))) # type: ignore[operator]
84 else:
85 msg = f"Unsafe node type: {type(node).__name__}"
86 raise ValueError(msg)
88 try:
89 tree = ast.parse(expression, mode="eval")
90 result = _eval(tree.body)
91 return result
92 except (ValueError, SyntaxError, ZeroDivisionError) as e:
93 msg = f"Invalid expression: {e}"
94 raise ValueError(msg) from e
97@tool
98def calculator(expression: str = Field(description="Mathematical expression to evaluate (e.g., '2 + 2', '10 * 3')")) -> str:
99 """
100 Evaluate a mathematical expression safely.
102 Supports: +, -, *, /, **, %, and parentheses.
103 Does NOT support: functions, variables, or imports for security.
105 Examples:
106 - "2 + 2" → 4.0
107 - "10 * 3 - 5" → 25.0
108 - "(8 / 2) ** 2" → 16.0
109 """
110 try:
111 _safe_log("info", "Calculator tool invoked", extra={"expression": expression})
112 _safe_metric("tool_calls", 1, {"tool": "calculator"})
114 result = _safe_eval(expression)
116 _safe_log("info", "Calculator result", extra={"expression": expression, "result": result})
117 return f"{result}"
119 except Exception as e:
120 error_msg = f"Error evaluating expression '{expression}': {e}"
121 _safe_log("error", error_msg, exc_info=True)
122 return f"Error: {e}"
125@tool
126def add(a: float = Field(description="First number"), b: float = Field(description="Second number")) -> str:
127 """Add two numbers together."""
128 try:
129 _safe_log("info", "Add tool invoked", extra={"a": a, "b": b})
130 _safe_metric("tool_calls", 1, {"tool": "add"})
132 result = a + b
133 return f"{result}"
135 except Exception as e:
136 _safe_log("error", f"Error in add tool: {e}", exc_info=True)
137 return f"Error: {e}"
140@tool
141def subtract(a: float = Field(description="First number"), b: float = Field(description="Second number")) -> str:
142 """Subtract second number from first number."""
143 try:
144 _safe_log("info", "Subtract tool invoked", extra={"a": a, "b": b})
145 _safe_metric("tool_calls", 1, {"tool": "subtract"})
147 result = a - b
148 return f"{result}"
150 except Exception as e:
151 _safe_log("error", f"Error in subtract tool: {e}", exc_info=True)
152 return f"Error: {e}"
155@tool
156def multiply(a: float = Field(description="First number"), b: float = Field(description="Second number")) -> str:
157 """Multiply two numbers together."""
158 try:
159 _safe_log("info", "Multiply tool invoked", extra={"a": a, "b": b})
160 _safe_metric("tool_calls", 1, {"tool": "multiply"})
162 result = a * b
163 return f"{result}"
165 except Exception as e:
166 _safe_log("error", f"Error in multiply tool: {e}", exc_info=True)
167 return f"Error: {e}"
170@tool
171def divide(a: float = Field(description="Numerator"), b: float = Field(description="Denominator (cannot be zero)")) -> str:
172 """Divide first number by second number."""
173 try:
174 _safe_log("info", "Divide tool invoked", extra={"a": a, "b": b})
175 _safe_metric("tool_calls", 1, {"tool": "divide"})
177 if b == 0:
178 return "Error: Division by zero"
180 result = a / b
181 return f"{result}"
183 except Exception as e:
184 _safe_log("error", f"Error in divide tool: {e}", exc_info=True)
185 return f"Error: {e}"