Repository navigation
Expand file tree
/
Copy pathreasoning_mcp_server.py
More file actions
executable file
路130 lines (110 loc) 路 4.73 KB
/
Copy pathreasoning_mcp_server.py
File metadata and controls
executable file
路130 lines (110 loc) 路 4.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# SPDX-FileCopyrightText: 2025 MiromindAI
#
# SPDX-License-Identifier: Apache-2.0
import os
from anthropic import Anthropic
from fastmcp import FastMCP
from openai import OpenAI
import asyncio
from src.logging.logger import setup_mcp_logging
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
ANTHROPIC_BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com")
ANTHROPIC_MODEL_NAME = os.environ.get(
"ANTHROPIC_MODEL_NAME", "claude-3-7-sonnet-20250219"
)
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
OPENAI_MODEL_NAME = os.environ.get("OPENAI_MODEL_NAME", "o3")
# Initialize FastMCP server
setup_mcp_logging(tool_name=os.path.basename(__file__))
mcp = FastMCP("reasoning-mcp-server")
@mcp.tool()
async def reasoning(question: str) -> str:
"""This tool is for pure text-based reasoning, analysis, and logical thinking. It integrates collected information, organizes final logic, and provides planning insights.
IMPORTANT: This tool cannot access the internet, read files, program, or process multimodal content. It only performs pure text reasoning.
Use this tool for:
- Integrating and synthesizing collected information
- Analyzing patterns and relationships in data
- Logical reasoning and problem-solving
- Planning and strategy development
- Complex math problems, puzzles, riddles, and IQ tests
DO NOT use this tool for simple and obvious questions.
Args:
question: The complex question or problem requiring step-by-step reasoning. Should include all relevant information needed to solve the problem.
Returns:
The reasoned answer to the question.
"""
messages_for_llm = [
{
"role": "user",
"content": [
{
"type": "text",
"text": question,
}
],
}
]
if OPENAI_API_KEY:
max_retries = 5
for attempt in range(1, max_retries + 1):
try:
client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
response = client.chat.completions.create(
model=OPENAI_MODEL_NAME,
messages=messages_for_llm,
extra_body={},
)
content = response.choices[0].message.content
# Check if content is empty and retry if so
if content and content.strip():
return content
else:
if attempt >= max_retries:
return f"Reasoning (OpenRouter Client) failed after {max_retries} retries: Empty response received\n"
await asyncio.sleep(
5 * (2**attempt)
) # Exponential backoff with max 30s
continue
except Exception as e:
if attempt >= max_retries:
return f"Reasoning (OpenRouter Client) failed after {max_retries} retries: {e}\n"
await asyncio.sleep(
5 * (2**attempt)
) # Exponential backoff with max 30s
else:
max_retries = 5
for attempt in range(1, max_retries + 1):
try:
client = Anthropic(
api_key=ANTHROPIC_API_KEY, base_url=ANTHROPIC_BASE_URL
)
response = client.messages.create(
model=ANTHROPIC_MODEL_NAME,
max_tokens=21000,
thinking={
"type": "enabled",
"budget_tokens": 19000,
},
messages=messages_for_llm,
stream=False,
)
content = response.content[-1].text
# Check if content is empty and retry if so
if content and content.strip():
return content
else:
if attempt >= max_retries:
return f"[ERROR]: Reasoning (Anthropic Client) failed after {max_retries} retries: Empty response received\n"
await asyncio.sleep(
5 * (2**attempt)
) # Exponential backoff with max 30s
continue
except Exception as e:
if attempt >= max_retries:
return f"[ERROR]: Reasoning (Anthropic Client) failed after {max_retries} retries: {e}\n"
await asyncio.sleep(
5 * (2**attempt)
) # Exponential backoff with max 30s
if __name__ == "__main__":
mcp.run(transport="stdio", show_banner=False)