MCP-Server

— AI-Agent Orchestrator Core [](https://python.org) [](https://ollama.com) [](#architecture) [](#protocol) [](#pipeline) **MCP Server** — an orchestrator core for launching and coordinating AI-agents, built on pure **Python 3.10+** without third-party frameworks. Each agent runs as an independent subprocess, communicating with the core via a simple JSON protocol over stdin/stdout. The core stores shared memory for agents, accesses the local LLM throughllama REST API**, supports live interactive dialogue, and can automatically user's request through the full **Analyst → Developer → Tester**, including automatic code fixing when. ## Table of Contents [Key Features](#key- [Project Structure](#project-structure [Implemented Components](#) - [Architecture](#architecture) - [Protocol](#protocol) - [Requirements](#requirements) - [Installation](#installation) - [ Check](#running-the-core- [Interactive Chat with Agent](#interactive-chat-with-agent) - [Analyst → Developer → Tester Chain](#analyst-%E2%86%92-developer-%E2%86%92-tester-chain) - [Task Specification Scheme and File System Security](#task-specification-scheme-and-file-system-security) - [Orchestrator Methods](#orchestrator-methods) - [Author](#author) ## Key Features ### Architecture - **Subprocess-agents** — each agent is a separate process (any language, not just Python), the core does not know about its internal implementation **Unified JSON protocol** — all messages between the core and agent are a single JSON string over stdin/stdout, without binding to a specific framework. - **Shared agent memory** — a simple key-value store in the core's memory with autos disk, through which the analyst specification to the developer, and the developer passes the files to the tester. -Blocking Read and Windows Encoding- Reading **stdout** and **stderr** of each agent is moved to separate threads (`threading.Thread` + `queue.Queue`) — the core does not "freeze" waiting for a slow agent's response, and agent diagnostics are immediately visible in the console with the prefix `[agent:<name>][stderr]`. - Agent subprocesses are forced `PYTHONIOING=utf-8`encoding="utf-8"` on reading — without this, on Windows (where the console often uses a legacy code page like cp1251), the agent would `UnicodeEncodeError` if the model generated any non-standard Unicode character (e.g., `│`). ### LLM Integration - Direct REST client to **Ollama** (`/api/generate`, `/api/tags`) without third-party SDKs — both at the core level (`Orchestrator.query_ollama`) and within agents. - No mocking (FakeLLM) — if Ollama is unavailable, not launched, or did not respond in time, the agent returns a clear, understandable error message. - Forced JSON mode for Ollama (`"format": "json"`) for task specification requests — forces the model to generate syntactically valid JSON at the decoding level, not just request" in the. ### Live Dialogue with Switching - `runyst.py` launches the and then allows rewriting with real-time. - Intent detection **at the code level**ants_spec()` in `run_analyst.py`): if the message contains the words "" / "technical specification"project plan" —ator itself enables strict validation mode, not relying solely on the model understanding it from the system prompt Full Analyst → Developer → Tester Cycle successful task specification validation, the orchestrator **automatically** launches the developer agent, passes the task specification to it, and records the generated code to- Immediately after that, the **tester agent** is automatically launched: generates unit tests for each file and runs them through `pytest If tests fail — the itself sends an error report developer with a request to fix the code **in the same project folder**, and runs the tests again. The cycle repeats until success or until the attempt limit is exhausted (default is 2) — without a single manual step. ### Observability - Logging of all key core events simultaneously to the console and to a file, broken down by date. - Separate log of invalid LLM responses — `logs/invalid_llm_responses.log`. - Separate log of file operations — `logs/fs_operations.log` (what was written, what was blocked, what failed with an error). - Separate log of pytest runs — `logs/test_runs.log` (how many tests passed/failed, return code). - When the developer's timeout is reached, the core itself looks into the project folder and shows what has already been written to disk — the agent process is not killed when the timeout is reached, it can continue working in the background. ## Project Structure ```text mcp_server/ ├── core/ │ ├── __init__.py │ ├── orchestrator.py # Orchestrator class + AgentHandle + build→test→fix cycle │ ├── parser.py # LLM response JSON validator (task specification scheme) │ ├── fs_tools.py # Safe write_file/create_directory │ └── mcp_tools.py # pytest launch + result parsing │ ├── agents/ │ ├── __init__.py │ ├── test_agent.py # Test agent stub (echo protocol) │ ├── analyst.py # AI-agent: conversational chat + task specification mode │ ├── developer.py # Developer agent: task specification → code files on disk │ └── tester.py # Tester agent: code → unit tests → pytest run │ ├── workspace/ │ ├── configs/ │ └── config.json # Ollama host, agent registry, timeouts │ ├── prompts/ │ │ ├── system_prompt.txt # Basic system prompt │ │ ├── analyst.md # Agent prompt (dialogue + strict JSON task specification) │ │ ├── developer.md # Developer prompt (only code, no markdown/explanations) │ │ └── tester.md # Tester prompt (only pytest test code) │ ├── rules/ │ │ └── rules.md # Protocol for core and agent communication │ └── projects/ # Developer and tester write files here │ ├── logs/ # Core logs, invalid LLM responses, file operations, and tests ├── run_test.py # Core check with test agent ├── run_analyst.py # Interactive chat + automatic full chain launch └── requirements.txt ``` ## Implemented Components ### ✅ Component 1 — `Orchestrator` (`core/orchestrator.py`) The core of the system, responsible for the entire lifecycle of agents, memory, and the Analyst→Developer→Tester chain. - Launch/stop agents as subprocesses (`start_agent`, `stop_agent`, `stop_all_agents`), with forced UTF-8 encoding. - Shared memory with persistence in `workspace/memory.json` (`remember`, ` `append_history`). - Receiving user input and routing first agent in the chain (`handle_user_input`, optionally with `parse_as_spec=True`). - `_run_pipeline(spec)` — high-level cycle: build → tests → (on failure) fix → tests again, with an attempt limit. - `_ensure_agent_running()` — common helper for launching/reusing any agent by file name. - Response timeouts for developer and tester scale number of files Diagnostics of a partially on timeout (`_report_partial_project`). client to Ollama: text generation and list of installedquery_ollama`, `list_ollama_models`). ### ✅ Component 2AgentHandle` (`core/orchestrator.py`) Internal wrapper around each launched agent. - Background threads for non-blocking reading of **stdout** (protocol responses) and **stderr** (, printed to the). - Message queue (`queue.Queue`) between the stdout reader thread and the core's main cycle. - Correct process termination with a fallback to `kill()`, if `terminate()` did not complete within the timeout. ### ✅ Component 3 — `test_agent` (`agents/test_agent.py`) Minimal stub agent for protocol testing — echo response with a message serial number. ### ✅ Component 4 — `analyst` (`agents/analyst.py`) Default - ordinary conversational assistant; on request ("draft ToR") - structured ToR generator. - Reads system prompt from `workspace/prompts/analyst.md`, host and model - from `workspace/configs/config.json`. - With `want_spec` flag from the core, passes `"format": "json"` to Ollama - forced JSON generation. - **No stubbing**: in case of connection error with Ollama, returns a real message with the reason (`ollama_unreachable`, `ollama_timeout`, `ollama_error`). - Works as a permanent process - handles messages one by one without restarting. ### ✅ Component 5 — `parser.py` (`core/parser.py`) Structured JSON response validator for LLM. - Extracts JSON object from response text (even from markdown block ```json``` or with explanations around). - **Normalizes field names**: if the model translated keys to Russian (`"Название"`, `"Описание"`, `"Требования"` etc. instead of `title`/`description`/`requirements`) - brings them to the expected scheme automatically before checking required fields. - Checks required fields (`title`, `description`, `requirements`, `modules`) and list types (`requirements`, `modules`, `tech_stack`, `files`). - Checks that each `files` element is an object with `path` field (needed for developer). - Logs each invalid response in `logs/invalid_llm_responses.log`. ### ✅ Component 6 — `developer` (`agents/developer.py`) Developer agent: turns ToR into real code files, can improve them according to the tester report. - Receives ToR either explicitly in the message (`{"type": "build_request", "spec": {...}}`), or reads it from `workspace/memory.json` as a reserve. - Iterates over `spec["files"]` list, for each file forms a contextual prompt (project name and description, requirements, stack, path and purpose of a specific file) and addresses Ollama. - **Supports `fix_request`**: when receiving `{"type": "fix_request", "project_dir": "...", "test_report": {...}}` regenerates files **in the same project folder**, adding to each file prompt pytest output and an explicit request to fix the failure reason without removing necessary functionality. - `clean_code_response()` removes markdown wrapper (```code```) and typical introductory phrases ("Here is the code:", "Of course, here it is:"), leaving only the code itself. - Transliterates project name into a safe folder name (Cyrillic → Latin) and protects against overwriting projects with the same name during the first build (adds a timestamp in case of collision). - Writes files through `core/fs_tools.py` - safely, in `workspace/projects/<slug>/`. ### ✅ Component 7 — `fs_tools.py` (`core/fs_tools.py`) MCP file system tools for developer and tester. - `write_file(relative_path, content)` and `create_directory(relative_path)` - both operations work strictly within `workspace/projects/`. - Protection against going out of bounds (`PathEscapeError`): blocks `../../etc/passwd`, absolute paths and any attempts to resolve a path beyond `workspace/projects/`. - Each operation (success/blocking/error) is logged in `logs/fs_operations.log`. ### ✅ Component 8 — `tester` (`agents/tester.py`) Tester agent: generates unit tests for developer's code and runs them. - Receives `project_dir` and list of generated files (`{"type": "test_request", "project_dir": "...", "files_written": [...]}`), or takes the last project from common memory as a reserve. - For each `.py` file reads source code and asks Ollama to generate tests based on `workspace/prompts/tester.md`. - Before test code, deterministically adds a header: `import pytest`, path to parent folder, **and** `import <module>`, **and** `from <module> import *` - so that both call styles work (`module.func()` and `func()` directly), as well as `pytest.raises`/fixtures. - Writes tests to `<project_dir>/tests/test_<module>.py` through `core/fs_tools.py`. - Runs `core/mcp_tools.run_pytest()` and returns to the orchestrator the number of passed/failed tests and the tail of pytest output. ### ✅ Component 9 — `mcp_tools.py` (`core/mcp_tools.py`) MCP tools for running tests. - `find_project_python()` - searches for an interpreter in the virtual environment of the generated project itself (`venv/`, `.venv/`), if the developer created it; otherwise uses the environment of MCP server itself (there is already `pytest` from `requirements.txt`). - `run_pytest()` - runs `python -m pytest` with a timeout, parses the final output string (`"N passed, M failed in X.XXs"`) into `passed`/`failed`/`collection_errors` counts. - Each run is logged in `logs/test_runs.log`. ## 🏗 Architecture ### Data flow: user → analyst → developer → tester → (fix) ``` ┌──────────────┐ handle_user_input() ┌────────────────┐ stdin (JSON) ┌──────────┐ │ User │─────────────────────►│ Orchestrator │───────────────►│ analyst │ └──────────────┘ │ │ │(subprocess)│ │ + memory.json │◄───────────────┤ │ └───────┬────────┘ stdout (JSON) └──────────┘ │ ToR is valid (parse_as_spec=true) ▼ ┌────────────────┐ build_request ┌──────────┐ │ _run_pipeline() │───────────────►│developer │ │ │◄───────────────┤ │ └───────┬────────┘ files_written └────┬─────┘ │ │ write_file() ▼ ▼ ┌────────────────┐ test_request workspace/projects/<slug>/ │ │───────────────►┌──────────┐ │ (same cycle) │◄───────────────┤ tester │ └───────┬────────┘ passed/failed └────┬─────┘ │ │ run_pytest() failed > 0 │ ▼ ▼ pytest report fix_request → developer (repeat, up to 2 attempts) ``` Dependencies go in one direction: **user → core → agent**. All three agents (analyst, developer, tester) address Ollama directly; the core can also address Ollama independently through `query_ollama()`. ### Why subprocesses and not threads/coroutines | Criterion | Subprocess agents | Threads in one process | |---|---|---| | **Crash isolation** | ✅ Agent crash doesn't kill the core | ❌ Uncaught exception can damage the process | | **Implementation language** | ✅ Agent can be in any language | ❌ Only Python | | **Protocol simplicity** | ✅ JSON string - universal interface | ⚠️ Need common Python objects | | **Parallelism** | ✅ Real OS parallelism | ⚠️ GIL limits pure Python | ## 🔌 Communication protocol between core and agent Core → agent (`stdin`): ```json {"type": "user_input", "text": "Draft ToR for reminders bot", "memory_snapshot": [...], "want_spec": true} ``` Analyst agent → core (`stdout`), ready ToR: ```json {"type": "agent_response", "agent": "analyst", "text": "{\"title\": \"...\", \"files\": [...]}", "source": "ollama"} ``` Core → developer agent (`stdin`), primary build: ```json {"type": "build_request", "spec": {"title": "...", "files": [{"path": "main.py", "description": "..."}]}} ``` Core → developer agent (`stdin`), fix request: ```json {"type": "fix_request", "spec": {...}, "project_dir": "bot_napominaniy", "test_report": {"passed": 1, "failed": 2, "raw_output": "..."}} ``` Developer agent → core (`stdout`): ```json {"type": "agent_response", "agent": "developer", "project_dir": "bot_napominaniy", "files_written": ["bot_napominaniy/main.py"], "errors": []} ``` Core → tester agent (`stdin`): ```json {"type": "test_request", "project_dir": "bot_napominaniy", "files_written": [...], "spec": {...}} ``` Tester agent → core (`stdout`): ```json {"type": "agent_response", "agent": "tester", "project_dir": "bot_napominaniy", "tests_written": ["bot_napominaniy/tests/test_main.py"], "passed": 3, "failed": 1, "raw_output": "..."} ``` Agent shutdown: ```json {"type": "shutdown"} ``` Complete list of rules - in [`workspace/rules/rules.md`](workspace/rules/rules.md). ## Requirements - Python 3.10+ - [Ollama](https://ollama.com) — installed and running locally (`ollama serve`) - At least one model downloaded in Ollama (e.g., `ollama pull llama3` or `ollama pull qwen2.5-coder:3b`) - `pytest` (automatically installed via `requirements.txt` — required for running tests of generated projects) --- ## Installation ```bash git clone https://github.com/yushmeh/MCP-Server cd MCP-Server pip install -r requirements.txt ``` Preparing Ollama: ```bash ollama serve ollama pull llama3 ``` Before the first real dialogue, it's recommended to "warm up" the model: ```bash ollama run llama3 >>> Hello # wait for a full response, then >>> /bye ``` --- ## Running Kernel Check ```bash python run_test.py ``` Runs `test_agent.py`, sends three test messages, shows overall memory, and attempts to connect to Ollama. --- ## Interactive Chat with Agent ```bash python run_analyst.py ``` Runs the agent once and opens a live dialogue. To exit — `exit`, `quit`, or `Ctrl+C`. ``` Agent is ready for dialogue. To exit, type: exit / quit You: hello Agent: Hello! How can I help you? You: exit ``` If Ollama is unavailable, the agent will return a real error message, and the dialogue can continue. --- ## Analyst → Developer → Tester Chain If the message contains the words **"тз"**, **"technical task"**, or **"project plan"** — `run_analyst.py` itself (at the code level, not relying on the model) enables strict mode: ``` You: Compose a technical specification for a Telegram bot to remind about medication intake 🛠 Launching developer agent for «...»... ✅ Generated 5 files in workspace/projects/<slug>/ 🧪 Launching tester agent for «<slug>»... Test result: 4 passed, 1 failed 🔁 Tests failed (1). Passing error report to developer with request «fix error» (attempt 1/2)... ✅ Generated 5 files in workspace/projects/<slug>/ 🧪 Launching tester agent for «<slug>»... Test result: 5 passed, 0 failed ✅ All tests passed successfully for «<slug>». Task completed. ``` What's happening behind the scenes (method `Orchestrator._run_pipeline`): 1. The orchestrator asks Ollama to generate a technical specification with `"format": "json"` (forced JSON), and `core/parser.py` validates and normalizes the response; 2. Upon success, the technical specification is saved to memory under the key `last_tech_spec`, and `developer.py` is launched — generates code according to `spec["files"]`; 3. Immediately, `tester.py` is launched — generates tests for each file and runs them through `core/mcp_tools.run_pytest()`; 4. If all tests pass — the pipeline completes with a success message; 5. If there are failures — the orchestrator sends a `fix_request` to the developer with the pytest output, and the developer regenerates files **in the same folder**, and the tester runs the tests again; 6. The cycle repeats until success or until `max_fix_attempts` (default 2) — after which the pipeline honestly stops with a message about the need for manual verification, without looping; 7. Timeouts for waiting for the developer and tester are scaled according to the number of files — generation on CPU without GPU can take 1-2 minutes per file; 8. If a timeout occurs at the build stage — the orchestrator looks into the project folder and shows what has already been written to disk (the agent process is not killed, may continue writing files in the background). To call the chain programmatically, bypassing the chat: ```python response = orch.handle_user_input("Compose a technical specification for ...", parse_as_spec=True) ``` --- ## Technical Specification Schema and File System Security Full JSON schema generated by the analyst and consumed by the developer and tester: ```json { "title": "Short project name", "description": "Detailed task description", "requirements": ["Requirement 1", "Requirement 2"], "modules": ["Module 1", "Module 2"], "tech_stack": ["Technology 1", "Technology 2"], "files": [ {"path": "main.py", "description": "Application entry point"}, {"path": "models.py", "description": "Data models"} ] } ``` `title`, `description`, `requirements`, `modules` are required; `files` is required for launching the developer (without it, a single `main.py` is used). All files written by the developer and tester go through `core/fs_tools.py`: - the path is resolved and checked to remain strictly within `workspace/projects/`; - absolute paths and `../../` outside the sandbox are blocked (`PathEscapeError`); - each operation is logged in `logs/fs_operations.log`. Tester tests are additionally run through `core/mcp_tools.run_pytest()` — on the best available interpreter (project venv or MCP server environment), with logging in `logs/test_runs.log`. --- ## Orchestrator Methods | Method | Purpose | |---|---| | `start_agent(name, script_path)` | Run agent as a subprocess (UTF-8 encoding) | | `stop_agent(name)` / `stop_all_agents()` | Stop one agent or all | | `handle_user_input(text, parse_as_spec=False)` | Pass input to the first agent; with `True` — validate as technical specification and launch full pipeline | | `remember(key, value)` / `recall(key)` | Write/read common agent memory | | `append_history(role, content)` | Add entry to common dialogue history | | `query_ollama(prompt, model=...)` | Request to local LLM via Ollama (at the kernel level) | | `list_ollama_models()` | List of installed Ollama models | | `_handle_analyst_response(response)` | Technical specification validation via `core.parser`, saving to memory, launching `_run_pipeline` | | `_run_pipeline(spec, max_fix_attempts=2)` | Full cycle: assembly → tests → fix → tests, until success or attempt limit | | `_ensure_agent_running(name, script)` | Launch/reuse any agent by file name | | `_trigger_developer(spec)` / `_trigger_developer_fix(...)` | Primary assembly / repeated assembly with error report | | `_trigger_tester(project_dir, files, spec)` | Launch tester for an already assembled project | | `_print_dev_result(response)` | Printing assembly result (files/errors) | | `_report_partial_project(spec)` | On timeout — show what's already written to disk | | `shutdown()` | Correctly stop all agents and save memory | --- ## Author [@yushmeh](https://github.com/yushmeh)