diff --git a/app/api.py b/app/api.py
index e4123eb..21bcb2b 100644
--- a/app/api.py
+++ b/app/api.py
@@ -10,6 +10,7 @@ from flask_cors import CORS
from flask_socketio import SocketIO
from workflows import graph_history_manager, run_agent_streaming
from llm_client import DEFAULT_TEMPERATURE, MODELS, get_llm # Добавляем импорт списка моделей
+import base64
api = Flask(__name__)
CORS(
@@ -260,6 +261,11 @@ def chat_stream():
# Запускаем стриминг ответа от LLM (используем обновленный агент)
accumulated_content = ""
+ # Вспомогательная функция для безопасной передачи JSON в HTML-атрибутах
+ def safe_b64(text):
+ if not text: return "e30=" # пустой json '{}'
+ return base64.b64encode(str(text).encode('utf-8')).decode('utf-8')
+
try:
for chunk_data in run_agent_streaming(graph_id, user_node_id,
assistant_node_id,
@@ -271,6 +277,27 @@ def chat_stream():
if chunk_data.get("type") in ["chunk", "tool_start", "tool_end"]:
if chunk_data.get("type") == "chunk":
accumulated_content += chunk_data.get("content", "")
+
+ elif chunk_data.get("type") == "tool_start":
+ # Встраиваем стартовый маркер в память базы данных
+ start_marker = f'\n\n\n\n'
+ accumulated_content += start_marker
+
+ elif chunk_data.get("type") == "tool_end":
+ # Находим стартовый маркер и меняем его на финальный с данными
+ search_marker = f''
+
+ req_b64 = safe_b64(chunk_data.get("request"))
+ res_b64 = safe_b64(chunk_data.get("response"))
+ status = "error" if chunk_data.get("is_error") else "success"
+
+ end_marker = f''
+
+ if search_marker in accumulated_content:
+ accumulated_content = accumulated_content.replace(search_marker, end_marker)
+ else:
+ accumulated_content += f'\n\n{end_marker}\n\n'
+
yield f"data: {json.dumps(chunk_data)}\n\n"
# Обработка ошибки во время стриминга (крашим узел на фронте)
diff --git a/app/mcp_tools.py b/app/mcp_tools.py
new file mode 100644
index 0000000..178e989
--- /dev/null
+++ b/app/mcp_tools.py
@@ -0,0 +1,125 @@
+from pydantic import BaseModel, Field, create_model
+from langchain_core.tools import StructuredTool
+
+# ----------------- MCP MANAGER -----------------
+try:
+ from mcp.client.stdio import stdio_client, StdioServerParameters
+ from mcp.client.session import ClientSession
+ MCP_AVAILABLE = True
+except ImportError:
+ MCP_AVAILABLE = False
+
+_MCP_CACHED_TOOLS = {}
+
+def json_schema_to_pydantic(schema: dict, model_name: str) -> type[BaseModel]:
+ """Преобразует JSON Schema от MCP сервера в Pydantic модель для LangChain"""
+ fields = {}
+ properties = schema.get("properties", {})
+ required = schema.get("required", [])
+
+ for key, val in properties.items():
+ t = val.get("type", "string")
+ py_type = str
+ if t == "integer": py_type = int
+ elif t == "number": py_type = float
+ elif t == "boolean": py_type = bool
+ elif t == "array": py_type = list
+ elif t == "object": py_type = dict
+
+ desc = val.get("description", "")
+ if key in required:
+ fields[key] = (py_type, Field(..., description=desc))
+ else:
+ fields[key] = (py_type, Field(default=None, description=desc))
+
+ if not fields:
+ fields["kwargs"] = (dict, Field(default_factory=dict, description="Аргументы"))
+
+ return create_model(model_name, **fields)
+
+def create_mcp_tool(server_config, tool_name, tool_desc, json_schema, full_env, args, debug_callback=None):
+ """Обертка, которая поднимает контейнер/процесс ровно на 1 вызов тула и убивает его"""
+ args_schema = json_schema_to_pydantic(json_schema, f"MCP_{tool_name.replace('-','_')}_Schema")
+ command = server_config.get("command")
+
+ def mcp_tool_runner(**kwargs):
+ import asyncio
+ import json
+ input_json = json.dumps(kwargs, indent=2, ensure_ascii=False)
+
+ async def _run():
+ server_params = StdioServerParameters(command=command, args=args, env=full_env)
+ async with stdio_client(server_params) as (read, write):
+ async with ClientSession(read, write) as session:
+ await session.initialize()
+ result = await session.call_tool(tool_name, arguments=kwargs)
+ text_outputs = []
+ for c in result.content:
+ if hasattr(c, 'text'):
+ text_outputs.append(c.text)
+ else:
+ text_outputs.append(str(c))
+ return "\n".join(text_outputs)
+
+ raw_output = asyncio.run(_run())
+
+ if debug_callback:
+ # debug_callback(f"MCP: {tool_name}\n" + input_json, raw_output)
+ debug_callback(kwargs, raw_output)
+
+ return f"Инструмент '{tool_name}' вернул следующий результат:\n\n{raw_output}\n"
+
+ return StructuredTool.from_function(
+ func=mcp_tool_runner,
+ name=tool_name.replace('-','_'), # Langchain не любит тире в именах
+ description=tool_desc or f"MCP Tool {tool_name}",
+ args_schema=args_schema
+ )
+
+def fetch_mcp_tools(server_config, debug_callback=None):
+ """Один раз читает список тулов от сервера и кеширует их схемы"""
+ if not MCP_AVAILABLE:
+ print("⚠️ MCP серверы настроены, но библиотека 'mcp' не установлена. Выполните: pip install mcp")
+ return []
+
+ import asyncio
+ import shlex
+ import os
+
+ config_hash = str(server_config) # primitive hash
+ if config_hash in _MCP_CACHED_TOOLS:
+ return _MCP_CACHED_TOOLS[config_hash]
+
+ async def _fetch():
+ command = server_config.get("command", "")
+ args_str = server_config.get("args", "")
+ args = shlex.split(args_str) if args_str else []
+
+ env_dict = {}
+ env_str = server_config.get("envString", "")
+ if env_str:
+ for pair in env_str.split(','):
+ if '=' in pair:
+ k, v = pair.split('=', 1)
+ env_dict[k.strip()] = v.strip()
+
+ full_env = {**os.environ.copy(), **env_dict}
+ server_params = StdioServerParameters(command=command, args=args, env=full_env)
+
+ async with stdio_client(server_params) as (read, write):
+ async with ClientSession(read, write) as session:
+ await session.initialize()
+ tools_resp = await session.list_tools()
+ return tools_resp.tools, full_env, args
+
+ try:
+ raw_tools, full_env, args = asyncio.run(_fetch())
+ wrapped_tools = []
+ for t in raw_tools:
+ wrapped_tools.append(create_mcp_tool(server_config, t.name, t.description, t.inputSchema, full_env, args, debug_callback))
+ _MCP_CACHED_TOOLS[config_hash] = wrapped_tools
+ print(f"🔌 Успешно загружено {len(wrapped_tools)} инструментов от MCP-сервера '{server_config.get('name')}'")
+ return wrapped_tools
+ except Exception as e:
+ print(f"❌ Ошибка инициализации MCP сервера {server_config.get('name')}: {e}")
+ return []
diff --git a/app/react_agent.py b/app/react_agent.py
index ec91bcf..8ee501c 100644
--- a/app/react_agent.py
+++ b/app/react_agent.py
@@ -16,9 +16,10 @@ import json
import os
import sys
from typing import Any, get_type_hints
-from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field, create_model
+from mcp_tools import fetch_mcp_tools
+
def build_react_agent(model_name: str, temperature: float, max_tokens: int, agency_mode: bool, obsidian_settings: dict, debug_callback=None):
"""Инициализация ядра на базе LangGraph."""
m_cfg = MODELS.get(model_name)
@@ -166,6 +167,8 @@ finally:
def get_dynamic_tools(obsidian_settings: dict, debug_callback=None):
"""Собирает инструменты двух видов на основе настроек Obsidian."""
tools = []
+
+ # Инициализация внутренних инструментов (скриптов)
custom_tools = obsidian_settings.get("customTools", [])
for ct in custom_tools:
@@ -247,6 +250,12 @@ def get_dynamic_tools(obsidian_settings: dict, debug_callback=None):
)
tools.append(help_tool)
+ # Инициализация внутренних MCP серверов
+ mcp_servers = obsidian_settings.get("mcpServers", [])
+ for mcp_server in mcp_servers:
+ mcp_tools = fetch_mcp_tools(mcp_server, debug_callback)
+ tools.extend(mcp_tools)
+
return tools
def _build_pydantic_schema_from_func(func) -> type[BaseModel]:
diff --git a/app/test_mcp.py b/app/test_mcp.py
new file mode 100644
index 0000000..d803e9d
--- /dev/null
+++ b/app/test_mcp.py
@@ -0,0 +1,26 @@
+# test_mcp.py
+import asyncio
+import sys
+print(f"Python: {sys.version}")
+
+from mcp.client.stdio import stdio_client, StdioServerParameters
+from mcp.client.session import ClientSession
+
+async def test():
+ params = StdioServerParameters(
+ command="podman",
+ args=["run", "-i", "--rm", "ddg-mcp-server"],
+ env=None
+ )
+ print("opening stdio_client...")
+ async with stdio_client(params) as (read, write):
+ print("creating session...")
+ async with ClientSession(read, write) as session:
+ print("initializing...")
+ await session.initialize()
+ print("OK!")
+ tools = await session.list_tools()
+ for t in tools.tools:
+ print(f" - {t.name}")
+
+asyncio.run(test())
\ No newline at end of file
diff --git a/app/workflows.py b/app/workflows.py
index d161315..2a6d504 100644
--- a/app/workflows.py
+++ b/app/workflows.py
@@ -105,6 +105,7 @@ def run_agent_streaming(graph_id: str,
# 4. Выполняем цикл с потоковой передачей событий
step_counter = 0
+ tool_mapping = {} # Хранит связку: ID вызова -> Номер шага
# LangGraph Stream Mode 'messages': отдает чанки токенов и вызовы функций
for chunk, metadata in agent_executor.stream({"messages": messages_for_llm}, stream_mode="messages"):
@@ -115,6 +116,10 @@ def run_agent_streaming(graph_id: str,
for tc in chunk.tool_call_chunks:
if "name" in tc and tc["name"]:
step_counter += 1
+ tc_id = tc.get("id", f"unknown_{step_counter}")
+ # Сохраняем шаг для этого конкретного вызова
+ tool_mapping[tc_id] = step_counter
+
yield {
"type": "tool_start",
"name": tc["name"],
@@ -132,25 +137,33 @@ def run_agent_streaming(graph_id: str,
# Тул отработал и принес ответ
elif hasattr(chunk, "type") and chunk.type == "tool":
+ req = ""
+ res = str(chunk.content)
+
+ # Достаем JSON отправленных параметров из последнего debug_callback
+ if current_run_debug:
+ dbg = current_run_debug.pop(0) # Берем первый из очереди
+ req = dbg.get("req", "")
+
+ # Ловушка для определения ошибок
+ is_error = False
+ if getattr(chunk, "status", "") == "error" or "ОШИБКА ИСПОЛНЕНИЯ" in res or "SCRIPT RUNTIME ERROR" in res:
+ is_error = True
+
+ # Достаем ИМЕННО ТОТ ШАГ, на котором начинался вызов этого инструмента
+ tc_id = getattr(chunk, "tool_call_id", "")
+ actual_step = tool_mapping.get(tc_id, step_counter) # Fallback, если id нет
+
yield {
"type": "tool_end",
- "name": chunk.name,
- "step": step_counter,
- "node_id": assistant_node_id
+ "name": getattr(chunk, "name", "tool"),
+ "step": actual_step,
+ "node_id": assistant_node_id,
+ "request": req,
+ "response": res,
+ "is_error": is_error
}
- # В САМОМ КОНЦЕ, после того как агент закончил работу (вышел из цикла)
- if current_run_debug and obsidian_settings.get('showDebugInfo'):
- debug_section = "\n\n---\n" + "\n\n---\n".join([
- f"**JSON Request:**\n```json\n{d['req']}\n```\n**Result:**\n```text\n{d['res']}\n```"
- for d in current_run_debug
- ])
- yield {
- "type": "chunk",
- "content": debug_section,
- "node_id": assistant_node_id
- }
-
except Exception as e:
print(f"Ошибка при стриминге: {e}")
import traceback