Aufbau lokaler KI-Systeme: Qwen3.6 + MCPs

# Python 3.11+ required
python --version

python -m venv qwen-mcp-env
supply qwen-mcp-env/bin/activate    # macOS / Linux
qwen-mcp-envScriptsactivate       # Home windows

# Core packages
pip set up 
  "openai>=1.30.0" 
  "qwen-agent>=0.0.10" 
  "mcp>=1.0.0" 
  "httpx>=0.27.0"

# Serving framework -- select one
pip set up "vllm>=0.19.0"       # NVIDIA GPU
pip set up "sglang>=0.5.10"     # NVIDIA GPU (sooner prefill for lengthy context)
pip set up "ktransformers"      # CPU/hybrid

# Node.js 18+ is required for pre-built MCP servers put in by way of npx
node --version

# Set up SGLang with full dependencies
pip set up "sglang(all)>=0.5.10"

# Serve Qwen3.6-35B-A3B with reasoning and tool-call parsers enabled.
# --reasoning-parser qwen3 appropriately handles the ... blocks.
# --tool-call-parser qwen3_coder routes software name outputs to the suitable format.
# --enable-prefix-caching is crucial for agent workloads -- permits KV cache reuse
#   throughout turns, which is what makes preserve_thinking environment friendly in follow.

python -m sglang.launch_server 
    --model-path Qwen/Qwen3.6-35B-A3B 
    --host 0.0.0.0 
    --port 30000 
    --reasoning-parser qwen3 
    --tool-call-parser qwen3_coder 
    --enable-prefix-caching 
    --tp 2    # tensor parallel throughout 2 GPUs; take away if utilizing single GPU

pip set up "vllm>=0.19.0"

# vLLM equal with the identical crucial flags
vllm serve Qwen/Qwen3.6-35B-A3B 
    --host 0.0.0.0 
    --port 8000 
    --reasoning-parser qwen3 
    --tool-call-parser qwen3_coder 
    --enable-prefix-caching-v2 
    --tensor-parallel-size 2

ollama pull qwen2.5:7b
ollama serve
# Ollama's API is OpenAI-compatible at http://localhost:11434/v1

# Well being examine -- ought to return {"standing": "okay"} or related
curl http://localhost:30000/well being

# Check the chat completions endpoint with a easy question
curl http://localhost:30000/v1/chat/completions 
  -H "Content material-Kind: utility/json" 
  -d '{
    "mannequin": "Qwen/Qwen3.6-35B-A3B",
    "messages": ({"position": "consumer", "content material": "Reply with: prepared"}),
    "max_tokens": 10
  }'

# Set your GitHub private entry token
# Required by the GitHub MCP server for API calls
export GITHUB_TOKEN=ghp_your_token_here

# Pre-built MCP servers set up by way of npx -- no separate set up step
# npx handles this on first use when the agent begins the servers
# Confirm npx is offered:
npx --version

mkdir qwen-github-agent
cd qwen-github-agent

# github_agent_qwenagent.py
# Conditions: pip set up qwen-agent openai
#   npm / npx have to be put in for the MCP servers
#   GITHUB_TOKEN env var have to be set
#   Native serving endpoint have to be operating (see earlier part)
#
# Find out how to run:
#   python github_agent_qwenagent.py

from qwen_agent.brokers import Assistant

# ── Server configuration ──────────────────────────────────────────────────────

# Level at your native serving endpoint.
# Change the base_url to match whichever server you began:
#   SGLang:  http://localhost:30000/v1
#   vLLM:    http://localhost:8000/v1
#   Ollama:  http://localhost:11434/v1
LLM_CONFIG = {
    "mannequin":     "Qwen/Qwen3.6-35B-A3B",
    "model_server": "http://localhost:30000/v1",
    "api_key":   "EMPTY",           # Native servers don't require an actual key

    # Pondering mode sampling params (from the official mannequin card greatest practices)
    "generate_cfg": {
        "temperature":       0.6,
        "top_p":             0.95,
        "top_k":             20,
        "min_p":             0.0,
        "thought_in_history": True,   # That is the preserve_thinking flag in Qwen-Agent
    },
}

# ── MCP server configuration ──────────────────────────────────────────────────
# Every server key names the server; the worth is the stdio launch command.
# Qwen-Agent begins every server as a subprocess and manages the MCP periods.

MCP_SERVERS = {
    "mcpServers": {
        "filesystem": {
            "command": "npx",
            "args": (
                "-y",
                "@modelcontextprotocol/server-filesystem",
                # Grant the agent entry to the present working listing
                # In manufacturing, limit to the precise repository path
                "."
            )
        },
        "github": {
            "command": "npx",
            "args": ("-y", "@modelcontextprotocol/server-github"),
            "env": {
                # The GitHub MCP server reads this env var for API authentication
                "GITHUB_TOKEN": "${GITHUB_TOKEN}"
            }
        },
    }
}

# ── System immediate ─────────────────────────────────────────────────────────────
SYSTEM_PROMPT = """You're a senior software program engineer with full entry to a GitHub repository
by way of MCP instruments.

When given a repository and process:
1. Checklist open points to grasp what wants fixing
2. Use filesystem instruments to learn related supply recordsdata and checks
3. Establish the basis trigger primarily based on the code and the difficulty description
4. Write a focused repair -- minimal modifications, no refactoring unrelated to the bug
5. Create a pull request with a transparent title and outline referencing the difficulty

All the time clarify your reasoning at every step. Suppose by edge instances earlier than writing code.
If you're unsure a few file's function, learn it earlier than modifying it."""

# ── Agent setup ───────────────────────────────────────────────────────────────

agent = Assistant(
    llm=LLM_CONFIG,
    identify="GitHub Developer Assistant",
    description="Reads points, fixes bugs, opens pull requests -- domestically by way of MCP.",
    system_message=SYSTEM_PROMPT,
    mcp_servers=MCP_SERVERS,
)

# ── Run the agent ─────────────────────────────────────────────────────────────

def run_agent(process: str):
    """
    Run the agent on a process description and stream the output.
    The agent will make software calls routinely; Qwen-Agent handles
    the total loop together with software execution and consequence injection.
    """
    messages = ({"position": "consumer", "content material": process})

    print(f"Process: {process}n{'─' * 70}")

    # Qwen-Agent's run() is a generator that yields intermediate steps
    # Every yielded message exhibits a software name, a software consequence, or the ultimate reply
    for response in agent.run(messages=messages):
        # response is a listing of messages representing the dialog to date
        # The final message comprises the newest output
        final = response(-1)
        position    = final.get("position", "")
        content material = final.get("content material", "")

        if position == "assistant" and content material:
            # Strip and show the considering block individually for readability
            import re
            considering = re.search(r"(.*?)", content material, re.DOTALL)
            if considering:
                print(f"(considering) {considering.group(1).strip()(:200)}...")
            clear = re.sub(r".*?", "", content material, flags=re.DOTALL).strip()
            if clear:
                print(f"(agent) {clear}")

        elif position == "software":
            tool_name = final.get("identify", "unknown_tool")
            print(f"(software:{tool_name}) consequence obtained")


if __name__ == "__main__":
    run_agent(
        "Within the repository myorg/my-api-project, discover the open subject about "
        "the login endpoint returning 200 for invalid tokens. Learn the related "
        "code and checks, repair the bug, and open a pull request."
    )

python github_agent_qwenagent.py

# github_agent_raw.py
# Conditions: pip set up mcp openai httpx
#   GITHUB_TOKEN env var have to be set, native server have to be operating
#
# Find out how to run:
#   python github_agent_raw.py

import asyncio
import json
import os
import re
from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.consumer.stdio import stdio_client

# ── Native serving consumer ───────────────────────────────────────────────────────
consumer = AsyncOpenAI(
    base_url="http://localhost:30000/v1",
    api_key="EMPTY",
)

MODEL = "Qwen/Qwen3.6-35B-A3B"

# ── Response processing ───────────────────────────────────────────────────────

def strip_thinking(textual content: str) -> str:
    """Take away ... blocks. Used after we solely want the motion."""
    return re.sub(r".*?", "", textual content, flags=re.DOTALL).strip()

def extract_thinking(textual content: str) -> str:
    """Extract the content material of the considering block for logging."""
    m = re.search(r"(.*?)", textual content, re.DOTALL)
    return m.group(1).strip() if m else ""

def process_response(response, preserve_thinking: bool = True) -> dict:
    """
    Course of a chat completion response from Qwen3.6.

    Handles two output codecs:
    1. Instrument name by way of the API's function_call / tool_calls area (when --tool-call-parser is energetic)
    2. Instrument name embedded within the message content material as JSON

    Args:
        response:          The OpenAI-compatible completion response
        preserve_thinking: If True, maintain considering content material in output for
                           the subsequent flip's KV cache profit

    Returns:
        dict with considering, tool_calls, final_answer, has_tool_calls, is_terminal
    """
    alternative  = response.selections(0)
    message = alternative.message

    # Path 1: Instrument calls within the structured area (most well-liked -- requires tool-call-parser flag)
    if message.tool_calls:
        tool_calls = (
            {
                "identify":      tc.operate.identify,
                "arguments": json.hundreds(tc.operate.arguments),
                "call_id":   tc.id,
            }
            for tc in message.tool_calls
        )
        considering = extract_thinking(message.content material or "")
        return {
            "considering":       considering if preserve_thinking else "",
            "tool_calls":     tool_calls,
            "final_answer":   "",
            "has_tool_calls": True,
            "is_terminal":    False,
        }

    # Path 2: Instrument calls embedded in content material textual content (fallback)
    content material = message.content material or ""
    tag_matches = re.findall(r"(.*?)", content material, re.DOTALL)
    tool_calls = ()
    for m in tag_matches:
        strive:
            tool_calls.append(json.hundreds(m.strip()))
        besides json.JSONDecodeError:
            go

    considering     = extract_thinking(content material)
    final_answer = re.sub(r".*?", "", content material, flags=re.DOTALL)
    final_answer = re.sub(r".*?", "", final_answer, flags=re.DOTALL).strip()

    return {
        "considering":       considering if preserve_thinking else "",
        "tool_calls":     tool_calls,
        "final_answer":   final_answer,
        "has_tool_calls": len(tool_calls) > 0,
        "is_terminal":    len(tool_calls) == 0 and bool(final_answer),
    }

# ── Core agent loop ───────────────────────────────────────────────────────────

async def run_github_agent(process: str, repo: str, max_turns: int = 20):
    """
    Run the GitHub developer assistant agent.

    Connects to filesystem and GitHub MCP servers, discovers their instruments,
    and runs the Qwen3.6 agent loop till the duty is full or max_turns reached.
    """
    # Begin each MCP servers and set up periods
    fs_params = StdioServerParameters(
        command="npx",
        args=("-y", "@modelcontextprotocol/server-filesystem", "."),
    )
    gh_params = StdioServerParameters(
        command="npx",
        args=("-y", "@modelcontextprotocol/server-github"),
        env={**os.environ, "GITHUB_TOKEN": os.environ.get("GITHUB_TOKEN", "")},
    )

    async with stdio_client(fs_params) as (fs_read, fs_write), 
               ClientSession(fs_read, fs_write) as fs_session, 
               stdio_client(gh_params) as (gh_read, gh_write), 
               ClientSession(gh_read, gh_write) as gh_session:

        # Initialize each periods
        await fs_session.initialize()
        await gh_session.initialize()

        # Uncover all out there instruments from each servers
        fs_tools_result = await fs_session.list_tools()
        gh_tools_result = await gh_session.list_tools()

        # Construct the OpenAI-format software record for the mannequin
        all_tools = ()
        tool_to_session = {}   # Maps software identify to the MCP session that owns it

        for software in fs_tools_result.instruments:
            all_tools.append({
                "kind": "operate",
                "operate": {
                    "identify":        software.identify,
                    "description": software.description,
                    "parameters":  software.inputSchema,
                }
            })
            tool_to_session(software.identify) = fs_session

        for software in gh_tools_result.instruments:
            all_tools.append({
                "kind": "operate",
                "operate": {
                    "identify":        software.identify,
                    "description": software.description,
                    "parameters":  software.inputSchema,
                }
            })
            tool_to_session(software.identify) = gh_session

        print(f"Instruments out there: {len(all_tools)} ({len(fs_tools_result.instruments)} filesystem, "
              f"{len(gh_tools_result.instruments)} GitHub)")

        # Construct dialog historical past
        system_prompt = f"""You're a senior software program engineer with entry to the repository {repo}.
Use the out there instruments to research points, learn code, write fixes, and create pull requests.
Suppose step-by-step. Learn earlier than you modify. Minimal modifications solely."""

        messages = (
            {"position": "system",  "content material": system_prompt},
            {"position": "consumer",    "content material": process},
        )

        # ── Agent loop ─────────────────────────────────────────────────────────
        for flip in vary(max_turns):
            print(f"n(Flip {flip + 1})")

            # Name the mannequin
            response = await consumer.chat.completions.create(
                mannequin=MODEL,
                messages=messages,
                instruments=all_tools if all_tools else None,
                tool_choice="auto",
                # Pondering mode sampling params from the official greatest practices
                temperature=0.6,
                top_p=0.95,
                top_k=20,
                min_p=0.0,
                max_tokens=4096,
                extra_body={
                    # preserve_thinking retains reasoning context throughout turns
                    # for KV cache effectivity on lengthy agent periods
                    "preserve_thinking": True,
                }
            )

            consequence = process_response(response, preserve_thinking=True)

            if consequence("considering"):
                print(f"(considering) {consequence('considering')(:200)}...")

            # Terminal state -- agent has produced a ultimate reply
            if consequence("is_terminal"):
                print(f"n(DONE)n{consequence('final_answer')}")
                return consequence("final_answer")

            # Instrument name state -- execute every software and inject outcomes
            if consequence("has_tool_calls"):
                # Append the assistant's message with software calls to historical past
                messages.append({
                    "position":       "assistant",
                    "content material":    response.selections(0).message.content material or "",
                    "tool_calls": response.selections(0).message.tool_calls or (),
                })

                for name in consequence("tool_calls"):
                    tool_name = name("identify")
                    tool_args = name.get("arguments", {})
                    call_id   = name.get("call_id", "")

                    print(f"(software) {tool_name}({json.dumps(tool_args)(:80)}...)")

                    session = tool_to_session.get(tool_name)
                    if not session:
                        result_content = f"Error: software '{tool_name}' not discovered"
                    else:
                        strive:
                            tool_result = await session.call_tool(tool_name, tool_args)
                            result_content = str(tool_result.content material)
                            # Truncate very lengthy outcomes to guard context price range
                            if len(result_content) > 12000:
                                result_content = result_content(:12000) + "n...(truncated)"
                        besides Exception as e:
                            result_content = f"Error: {e}"

                    print(f"(consequence) {result_content(:150)}...")

                    messages.append({
                        "position":        "software",
                        "content material":     result_content,
                        "tool_call_id": call_id,
                        "identify":        tool_name,
                    })

        print(f"(WARNING) max_turns ({max_turns}) reached with out terminal state")


# ── Entry level ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    asyncio.run(run_github_agent(
        process=(
            "Discover the open subject concerning the login endpoint returning 200 for invalid tokens. "
            "Learn src/auth.py and checks/test_auth.py to grasp the bug. "
            "Repair the verify_token operate and open a pull request together with your modifications."
        ),
        repo="myorg/my-api-project",
    ))

python github_agent_raw.py

# code_quality_server.py
# A customized MCP server exposing code high quality instruments to Qwen3.6.
#
# Conditions:
#   pip set up mcp ruff pytest
#
# Find out how to run standalone (for testing):
#   python code_quality_server.py
#
# So as to add to the Qwen-Agent config:
#   "code_quality": {
#       "command": "python",
#       "args": ("/absolute/path/to/code_quality_server.py")
#   }

import asyncio
import json
import subprocess
import sys
from mcp.server.fastmcp import FastMCP

# FastMCP is a high-level MCP server framework -- reduces boilerplate considerably
mcp = FastMCP("code_quality")


@mcp.software()
def run_linter(file_path: str, repair: bool = False) -> str:
    """
    Run ruff linter on a Python file and return structured lint outcomes.
    Use this earlier than modifying a file to grasp its present high quality state,
    and after making modifications to confirm the repair didn't introduce new points.

    Args:
        file_path: Absolute or relative path to the Python file to lint.
        repair:       If true, routinely repair secure points in place.

    Returns:
        JSON string with points record, subject rely, and recordsdata modified.
    """
    cmd = ("python", "-m", "ruff", "examine", file_path, "--output-format=json")
    if repair:
        cmd.append("--fix")

    strive:
        consequence = subprocess.run(cmd, capture_output=True, textual content=True, timeout=30)
        # ruff returns exit code 1 when points are discovered -- not an error
        output = consequence.stdout or consequence.stderr

        # Parse ruff's JSON output
        strive:
            points = json.hundreds(output) if output.strip() else ()
        besides json.JSONDecodeError:
            points = ()

        formatted = (
            {
                "line":    subject.get("location", {}).get("row", 0),
                "col":     subject.get("location", {}).get("column", 0),
                "code":    subject.get("code", ""),
                "message": subject.get("message", ""),
                "fix_available": subject.get("repair") will not be None,
            }
            for subject in points
            if isinstance(subject, dict)
        )

        return json.dumps({
            "file":         file_path,
            "points":       formatted,
            "total_issues": len(formatted),
            "fastened":        "auto-fix utilized" if repair else "no auto-fix",
        }, indent=2)

    besides subprocess.TimeoutExpired:
        return json.dumps({"error": "Linter timed out after 30s", "file": file_path})
    besides FileNotFoundError:
        return json.dumps({"error": "ruff not discovered -- set up with: pip set up ruff"})


@mcp.software()
def run_tests(goal: str, verbose: bool = False) -> str:
    """
    Run pytest on a module or listing and return structured go/fail outcomes.
    Use this after writing a repair to confirm the repair makes failing checks go
    with out breaking different checks.

    Args:
        goal:  Path to the take a look at file or listing to run (e.g. checks/, checks/test_auth.py)
        verbose: If true, embody full pytest output within the consequence.

    Returns:
        JSON string with go rely, fail rely, failure particulars, and period.
    """
    cmd = ("python", "-m", "pytest", goal, "--json-report", "--json-report-file=-", "-q")
    if verbose:
        cmd.append("-v")

    strive:
        consequence = subprocess.run(cmd, capture_output=True, textual content=True, timeout=120)
        output = consequence.stdout

        # Parse pytest-json-report output if out there
        strive:
            report = json.hundreds(output)
            abstract  = report.get("abstract", {})
            failures = (
                {
                    "take a look at":    t("nodeid"),
                    "message": t.get("name", {}).get("longrepr", "")(:500),
                }
                for t in report.get("checks", ())
                if t.get("end result") == "failed"
            )
            return json.dumps({
                "goal":   goal,
                "handed":   abstract.get("handed", 0),
                "failed":   abstract.get("failed", 0),
                "errors":   abstract.get("error", 0),
                "complete":    abstract.get("complete", 0),
                "period": abstract.get("period", 0),
                "failures": failures,
                "stdout":   consequence.stdout(:2000) if verbose else "",
            }, indent=2)

        besides json.JSONDecodeError:
            # Fallback: return uncooked output if JSON report not out there
            return json.dumps({
                "goal":  goal,
                "stdout":  consequence.stdout(:3000),
                "stderr":  consequence.stderr(:1000),
                "exit_code": consequence.returncode,
            })

    besides subprocess.TimeoutExpired:
        return json.dumps({"error": f"Exams timed out after 120s for goal: {goal}"})
    besides FileNotFoundError:
        return json.dumps({"error": "pytest not discovered -- set up with: pip set up pytest"})


if __name__ == "__main__":
    mcp.run(transport="stdio")

# In Qwen-Agent MCP_SERVERS dict:
"code_quality": {
    "command": "python",
    "args": ("/absolute/path/to/code_quality_server.py")
}

# Within the uncooked SDK, add a 3rd StdioServerParameters:
cq_params = StdioServerParameters(
    command="python",
    args=("/absolute/path/to/code_quality_server.py"),
)

# Check the server in MCP inspector mode
npx @modelcontextprotocol/inspector python code_quality_server.py
# Opens a browser UI the place you may name run_linter and run_tests immediately

# Pondering mode (planning, debugging, complicated multi-file duties)
THINKING_PARAMS = {
    "temperature": 0.6,
    "top_p":       0.95,
    "top_k":       20,
    "min_p":       0.0,
}

# Non-thinking mode (mechanical loops, quick standing checks)
# Cross enable_thinking=False within the chat template, or use system immediate:
# Add "/no_think" to the system immediate to suppress considering mode.
NON_THINKING_PARAMS = {
    "temperature": 0.7,
    "top_p":       0.8,
    "top_k":       20,
    "min_p":       0.0,
}