API Reference

Generated from docstrings with mkdocstrings.

Top-level package

mcp_for_copilot

mcp-for-copilot — an MCP gateway for GPT-6 Sol and friends.

Two ways to use this package:

As a library — wire the layers yourself::

from mcp_for_copilot import Gateway

gw = Gateway(api_key="sk-...", model="gpt-6-sol")
result = await gw.chat("Explain MCP in one paragraph.")

As a server — run the OpenAI-compatible FastAPI app::

mcp-for-copilot serve --port 8000

Layered architecture (each layer is independently testable):

=================== ================================================== config Configuration — env-driven, no hardcoded secrets logging_utils Logging — redacts secrets before they are emitted provider LLM provider adapter — the only place that calls a model mcp_client MCP client — stdio / http / sse transports tool_router Tool router — allowlist + argument validation approval Approval layer — gates risky tools orchestrator Orchestrator — wires the layers, caps tool rounds gateway High-level facade + OpenAI-compatible server =================== ==================================================

ApprovalDecision dataclass

ApprovalDecision(
    approved: bool, reason: str = "", approver: str = ""
)

The approver's verdict.

ApprovalLayer

ApprovalLayer(approver: Approver | None = None)

Wraps an approver callable and enforces fail-closed behaviour.

Source code in src/mcp_for_copilot/approval.py

def __init__(self, approver: Approver | None = None) -> None:
    self._approver = approver

request async

request(request: ApprovalRequest) -> ApprovalDecision

Ask the approver. Denies when no approver is configured.

Source code in src/mcp_for_copilot/approval.py

async def request(self, request: ApprovalRequest) -> ApprovalDecision:
    """Ask the approver. Denies when no approver is configured."""
    if self._approver is None:
        log_warning(
            "approval denied for %s: no approver configured (fail closed)",
            redact(request.tool),
        )
        return ApprovalDecision.deny(
            "no approver configured; risky tools are denied by default",
            approver="none",
        )

    try:
        decision = await self._approver(request)
    except Exception as exc:
        log_warning("approver raised for %s: %s", redact(request.tool), type(exc).__name__)
        return ApprovalDecision.deny(f"approver raised {type(exc).__name__}", approver="error")

    if not isinstance(decision, ApprovalDecision):
        return ApprovalDecision.deny(
            "approver returned an invalid decision", approver="invalid"
        )

    log_info(
        "approval %s for %s by %s",
        "granted" if decision.approved else "denied",
        redact(request.tool),
        redact(decision.approver or "unknown"),
    )
    return decision

ApprovalRequest dataclass

ApprovalRequest(
    tool: str,
    arguments: dict[str, Any] = dict(),
    reason: str = "",
    round_index: int = 0,
)

What the orchestrator asks the approver to decide on.

Gateway

Gateway(
    *,
    settings: Settings | None = None,
    provider: LLMProvider | None = None,
    approver: Approver | None = None,
    connect_mcp: bool = True,
)

Owns the provider + MCP client + orchestrator for one process.

Source code in src/mcp_for_copilot/gateway/facade.py

def __init__(
    self,
    *,
    settings: Settings | None = None,
    provider: LLMProvider | None = None,
    approver: Approver | None = None,
    connect_mcp: bool = True,
) -> None:
    self.settings = settings or get_settings()
    self.provider = provider or LLMProvider()
    self.debug_marathon = DebugMarathon(self.provider)
    self._approver = approver
    self._connect_mcp = connect_mcp

    # ``client`` is either a single :class:`MCPClient` (deprecated
    # ``MCP_SERVER_URL`` path) or an :class:`MCPClientManager` when
    # ``MCP_SERVERS`` lists more than one server. The orchestrator already
    # branches on the type, so both are valid here.
    self.client: MCPClient | MCPClientManager | None = None
    self.router = ToolRouter(
        allowed_tools=self.settings.mcp_allowed_tools or None,
        require_approval=self.settings.mcp_require_approval,
    )
    self.approval = ApprovalLayer(approver)
    self.orchestrator = LLMMCPOrchestrator(
        provider=self.provider,
        client=None,
        router=self.router,
        approval=self.approval,
        settings=self.settings,
    )
    self._connected = False

chat async

chat(
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> GatewayResult

Run one chat turn through the full tool loop.

Source code in src/mcp_for_copilot/gateway/facade.py

async def chat(
    self,
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> GatewayResult:
    """Run one chat turn through the full tool loop."""
    if not self._connected:
        await self.start()
    result = await self.orchestrator.run(
        messages, system=system, model_id=model_id, max_rounds=max_rounds
    )
    return GatewayResult.from_orchestrator(result)

list_tools async

list_tools() -> list[dict[str, Any]]

Return the allowed tools as OpenAI function schemas.

Source code in src/mcp_for_copilot/gateway/facade.py

async def list_tools(self) -> list[dict[str, Any]]:
    """Return the allowed tools as OpenAI function schemas."""
    if not self._connected:
        await self.start()
    if self.client is None:
        return []
    try:
        if isinstance(self.client, MCPClientManager):
            listed = await self.client.list_all_tools()
            tools = [tool for tools in listed.values() for tool in tools]
        else:
            tools = await self.client.list_tools()
    except MCPClientError as exc:
        log_warning("could not list tools: %s", redact(str(exc)))
        return []
    self.router.register_all(tools)
    return LLMProvider.to_openai_tools(self.router.available())

start async

start() -> None

Connect to the MCP server(s), if any are configured.

MCP_SERVERS (a JSON array) takes precedence and yields an :class:MCPClientManager. When it is unset, the deprecated single MCP_SERVER_URL fields are used exactly as before, so existing deployments keep working unchanged.

Source code in src/mcp_for_copilot/gateway/facade.py

async def start(self) -> None:
    """Connect to the MCP server(s), if any are configured.

    ``MCP_SERVERS`` (a JSON array) takes precedence and yields an
    :class:`MCPClientManager`. When it is unset, the deprecated single
    ``MCP_SERVER_URL`` fields are used exactly as before, so existing
    deployments keep working unchanged.
    """
    if self._connected or not self._connect_mcp:
        return

    if self.settings.mcp_servers_parsed:
        self.client = await self._connect_manager()
    else:
        self.client = await self._connect_single()

    self.orchestrator.client = self.client
    self._connected = True

status

status() -> dict[str, Any]

Non-secret status snapshot, safe to return over HTTP.

Source code in src/mcp_for_copilot/gateway/facade.py

def status(self) -> dict[str, Any]:
    """Non-secret status snapshot, safe to return over HTTP."""
    if isinstance(self.client, MCPClientManager):
        servers = sorted(self.client._clients)
        server_info: dict[str, Any] = {"servers": servers}
        transport = "multi"
    else:
        servers = []
        server_info = self.client.server_info if self.client else {}
        transport = self.settings.mcp_transport
    return {
        "llm": self.provider.describe(),
        "mcp": {
            "transport": transport,
            "configured": self.settings.mcp_configured,
            "connected": self.client is not None,
            "server_info": server_info,
            "servers": servers,
            "allowed_tools": list(self.router.allowed_tools),
            "require_approval": self.settings.mcp_require_approval,
            "approver_configured": self.approval.configured,
        },
    }

GatewayResult dataclass

GatewayResult(
    content: str = "",
    model: str = "",
    rounds: int = 0,
    used_tools: list[str] = list(),
    invocations: list[dict[str, Any]] = list(),
    usage: dict[str, Any] = dict(),
    error: str = "",
    plan: list[dict[str, str]] = list(),
    plan_progress: dict[str, int] = dict(),
    loop_stopped: bool = False,
)

A chat completion plus the tool audit trail.

LLMMCPOrchestrator

LLMMCPOrchestrator(
    *,
    provider: LLMProvider | None = None,
    client: MCPClient | MCPClientManager | None = None,
    router: ToolRouter | None = None,
    approval: ApprovalLayer | None = None,
    settings: Settings | None = None,
)

Drives the model/tool loop with the three gates in place.

Source code in src/mcp_for_copilot/orchestrator.py

def __init__(
    self,
    *,
    provider: LLMProvider | None = None,
    client: MCPClient | MCPClientManager | None = None,
    router: ToolRouter | None = None,
    approval: ApprovalLayer | None = None,
    settings: Settings | None = None,
) -> None:
    self.settings = settings or get_settings()
    self.provider = provider or LLMProvider()
    self.client = client
    self.router = router or ToolRouter(
        allowed_tools=self.settings.mcp_allowed_tools or None,
        require_approval=self.settings.mcp_require_approval,
    )
    self.approval = approval or ApprovalLayer()
    # The model's self-reported plan for the current turn. Reset at the
    # start of every ``run()`` so one turn never inherits another's plan.
    self._plan: list[PlanStep] = []
    # Repeated-call detection for the current turn. Also reset per turn.
    self._loop = LoopDetector(
        soft_threshold=self.settings.mcp_loop_soft_threshold,
        hard_threshold=self.settings.mcp_loop_hard_threshold,
    )

run async

run(
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> OrchestratorResult

Run the loop and return the final answer with its audit trail.

Source code in src/mcp_for_copilot/orchestrator.py

async def run(
    self,
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> OrchestratorResult:
    """Run the loop and return the final answer with its audit trail."""
    rounds_cap = max_rounds if max_rounds is not None else self.settings.mcp_max_tool_rounds
    self._plan = []
    self._loop = LoopDetector(
        soft_threshold=self.settings.mcp_loop_soft_threshold,
        hard_threshold=self.settings.mcp_loop_hard_threshold,
    )
    history: list[dict[str, Any]] = [
        {"role": "system", "content": self._system_prompt(system)},
        *messages,
    ]

    result = OrchestratorResult(model=self.provider.model_id)
    tools = await self._available_tool_schemas()

    for round_index in range(rounds_cap + 1):
        result.rounds = round_index + 1
        try:
            response = await self.provider.complete(
                history, tools=tools or None, model_id=model_id
            )
        except LLMProviderError as exc:
            log_warning("orchestrator provider failure: %s", redact(str(exc)))
            result.error = str(exc)
            result.content = f"Provider error: {exc}"
            return result

        result.model = response.model or result.model
        if response.usage:
            result.usage = response.usage

        if not response.wants_tools:
            result.content = response.content
            return result

        if round_index >= rounds_cap:
            log_warning("tool round cap reached (%d)", rounds_cap)
            result.content = (
                response.content
                or f"Stopped after {rounds_cap} tool rounds without a final answer."
            )
            return result

        history.append(self._assistant_tool_message(response))

        for call in response.tool_calls:
            if call.name == PLAN_TOOL_NAME:
                invocation, message = self._handle_plan_call(call)
            else:
                invocation, message = await self._handle_tool_call(call, round_index)
            result.invocations.append(invocation)
            history.append(message)

            # Loop detection: only real tool calls count. The synthetic
            # plan tool is excluded because re-publishing a plan is normal.
            if self.settings.mcp_loop_detection and call.name != PLAN_TOOL_NAME:
                warn, stop, count = self._loop.record(call.name, call.arguments)
                if stop:
                    log_warning(
                        "loop detected: %s repeated %d times; stopping turn",
                        redact(call.name),
                        count,
                    )
                    result.loop_stopped = True
                    result.plan = list(self._plan)
                    result.content = (
                        response.content
                        or f"Stopped: the model repeated the same tool call "
                        f"({call.name}) {count} times without making progress."
                    )
                    return result
                if warn:
                    log_info("loop warning: %s repeated %d times", redact(call.name), count)
                    history.append(
                        {
                            "role": "system",
                            "content": _LOOP_WARNING.format(count=count),
                        }
                    )

        result.plan = list(self._plan)
        reminder = self._plan_message()
        if reminder is not None:
            history.append(reminder)

    return result

LLMProvider

LLMProvider(
    *,
    provider: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    model_id: str | None = None,
    timeout: float | None = None,
    transport: AsyncBaseTransport | None = None,
    fallback_api_key: str | None = None,
    fallback_base_url: str | None = None,
    fallback_model_id: str | None = None,
    fallback_enabled: bool | None = None,
)

OpenAI-compatible chat-completions adapter.

Only openai is implemented. An unknown LLM_PROVIDER raises rather than falling back, so a misconfiguration is never masked.

Source code in src/mcp_for_copilot/provider.py

def __init__(
    self,
    *,
    provider: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    model_id: str | None = None,
    timeout: float | None = None,
    transport: httpx.AsyncBaseTransport | None = None,
    fallback_api_key: str | None = None,
    fallback_base_url: str | None = None,
    fallback_model_id: str | None = None,
    fallback_enabled: bool | None = None,
) -> None:
    settings = get_settings()
    self.provider = (provider or settings.llm_provider or "").strip().lower()
    if self.provider not in SUPPORTED_PROVIDERS:
        raise LLMProviderError(
            f"Unsupported LLM_PROVIDER {redact(self.provider)!r}. "
            f"Supported: {', '.join(SUPPORTED_PROVIDERS)}"
        )
    self.api_key = (api_key if api_key is not None else settings.llm_api_key).strip()
    self.base_url = (
        (base_url if base_url is not None else settings.llm_base_url).strip().rstrip("/")
    )
    self.model_id = resolve_model_id(model_id or settings.llm_model_id or None)
    self.timeout = float(timeout if timeout is not None else settings.llm_timeout_seconds)
    # Injectable transport so tests never touch the network.
    self._transport = transport

    # -- Z.ai (GLM) backup provider -----------------------------------
    # Used only when the primary request fails with a transport error,
    # timeout, or 5xx. A 4xx is a caller/request problem, not an outage,
    # so it is never retried against the fallback.
    self.fallback_enabled = (
        settings.llm_fallback_enabled if fallback_enabled is None else fallback_enabled
    )
    self.fallback_api_key = (
        fallback_api_key if fallback_api_key is not None else settings.llm_fallback_api_key
    ).strip()
    self.fallback_base_url = (
        fallback_base_url
        if fallback_base_url is not None
        else settings.llm_fallback_base_url
    ).strip().rstrip("/")
    self.fallback_model_id = (
        fallback_model_id
        if fallback_model_id is not None
        else settings.llm_fallback_model_id
    ).strip()
    self.fallback_timeout = float(settings.llm_fallback_timeout_seconds)

fallback_configured property

fallback_configured: bool

True when the Z.ai backup is usable (enabled + key + model).

complete async

complete(
    messages: list[dict[str, Any]],
    *,
    tools: list[dict[str, Any]] | None = None,
    model_id: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    reasoning_effort: str | None = None,
) -> ProviderResponse

Send messages to the provider and return a normalised reply.

tools must already be in OpenAI function-calling format; the adapter does not transform or extend it.

When the primary provider fails with a transport error, timeout, or 5xx and a Z.ai fallback is configured, the same request is retried once against the fallback endpoint. A 4xx is never retried: it is a request problem, not an outage.

Source code in src/mcp_for_copilot/provider.py

async def complete(
    self,
    messages: list[dict[str, Any]],
    *,
    tools: list[dict[str, Any]] | None = None,
    model_id: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    reasoning_effort: str | None = None,
) -> ProviderResponse:
    """Send *messages* to the provider and return a normalised reply.

    ``tools`` must already be in OpenAI function-calling format; the
    adapter does not transform or extend it.

    When the primary provider fails with a transport error, timeout, or
    5xx and a Z.ai fallback is configured, the same request is retried
    once against the fallback endpoint. A 4xx is never retried: it is a
    request problem, not an outage.
    """
    if not self.api_key:
        raise LLMProviderError(
            "LLM_API_KEY is not configured. Set it in the environment (never in source)."
        )

    model = resolve_model_id(model_id or self.model_id)

    # A client-facing tier id (gpt-6-luna / gpt-6-sol / gpt-6-astra) is not
    # a real DeepSeek model. Translate it to deepseek-flash + the tier's
    # reasoning_effort before anything is sent upstream, otherwise DeepSeek
    # rejects the request with HTTP 400.
    #
    # This translation is ONLY valid when the upstream actually IS
    # DeepSeek. gpt-6-luna / gpt-6-sol / gpt-6-astra are also real OpenAI
    # model ids, so applying the table unconditionally rewrites them to
    # `deepseek-flash` and sends that to OpenAI, which answers
    #   404 "The model `deepseek-flash` does not exist"
    # and the gateway surfaces a 502. Gate on the provider (and on the
    # base_url as a belt-and-braces check for a DeepSeek-compatible host).
    if self._uses_deepseek():
        tier = resolve_deepseek_tier(model)
        if tier is not None:
            model = tier["model"]
            if reasoning_effort is None:
                reasoning_effort = tier["reasoning_effort"]
            if max_tokens is None:
                max_tokens = tier.get("max_tokens")

    body = self._build_body(
        model,
        messages,
        tools=tools,
        max_tokens=max_tokens,
        temperature=temperature,
        reasoning_effort=reasoning_effort,
    )

    # Route glm-* to Z.ai when the primary upstream is not Z.ai -- the
    # primary would 404 on every glm id and 4xx never hits the fallback.
    glm_route = self._glm_route() if model.startswith("glm-") else None
    if glm_route is not None:
        route_url, route_key, route_timeout = glm_route
        log_debug(
            "llm request provider=zai-primary-override model=%s messages=%d tools=%d",
            model,
            len(messages),
            len(tools or []),
        )
        return await self._post(
            base_url=route_url,
            api_key=route_key,
            body=body,
            timeout=route_timeout,
        )

    log_debug(
        "llm request provider=%s model=%s messages=%d tools=%d",
        self.provider,
        model,
        len(messages),
        len(tools or []),
    )

    try:
        return await self._post(
            base_url=self.base_url,
            api_key=self.api_key,
            body=body,
            timeout=self.timeout,
        )
    except LLMProviderError as primary_error:
        if not self.fallback_configured or not self._is_retryable(primary_error):
            raise
        log_warning(
            "primary LLM failed (%s); retrying on Z.ai fallback model=%s",
            redact(str(primary_error)),
            self.fallback_model_id,
        )
        return await self._complete_fallback(
            messages,
            tools=tools,
            max_tokens=max_tokens,
            temperature=temperature,
            reasoning_effort=reasoning_effort,
            primary_error=primary_error,
        )

describe

describe() -> dict[str, Any]

Non-secret description of the adapter, safe to return over HTTP.

Source code in src/mcp_for_copilot/provider.py

def describe(self) -> dict[str, Any]:
    """Non-secret description of the adapter, safe to return over HTTP."""
    return {
        "provider": self.provider,
        "model_id": self.model_id,
        "base_url": self.base_url,
        "api_key_present": bool(self.api_key),
        "timeout_seconds": self.timeout,
        "fallback_enabled": self.fallback_enabled,
        "fallback_configured": self.fallback_configured,
        "fallback_base_url": self.fallback_base_url,
        "fallback_model_id": self.fallback_model_id,
        "fallback_api_key_present": bool(self.fallback_api_key),
        "known_models": sorted(KNOWN_MODELS),
    }

to_openai_tools staticmethod

to_openai_tools(tools: list[Any]) -> list[dict[str, Any]]

Convert MCP tool descriptors into OpenAI function-calling schemas.

Accepts objects exposing name / description / input_schema (i.e. :class:mcp_for_copilot.mcp_client.MCPTool).

Source code in src/mcp_for_copilot/provider.py

@staticmethod
def to_openai_tools(tools: list[Any]) -> list[dict[str, Any]]:
    """Convert MCP tool descriptors into OpenAI function-calling schemas.

    Accepts objects exposing ``name`` / ``description`` / ``input_schema``
    (i.e. :class:`mcp_for_copilot.mcp_client.MCPTool`).
    """
    converted: list[dict[str, Any]] = []
    for tool in tools:
        name = getattr(tool, "name", None)
        if not name:
            continue
        schema = getattr(tool, "input_schema", None) or {
            "type": "object",
            "properties": {},
        }
        converted.append(
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": getattr(tool, "description", "") or "",
                    "parameters": schema,
                },
            }
        )
    return converted

LLMProviderError

Bases: RuntimeError

Raised for any provider failure. The message is always redacted.

MCPCallResult dataclass

MCPCallResult(
    tool: str,
    content: str = "",
    is_error: bool = False,
    raw: dict[str, Any] = dict(),
)

The outcome of one tools/call.

MCPClient

MCPClient(
    *,
    url: str,
    transport: str = "stdio",
    auth_token: str = "",
    timeout: float = 60.0,
    env: dict[str, str] | None = None,
    transport_impl: AsyncBaseTransport | None = None,
)

Async MCP client for stdio, http and sse transports.

Source code in src/mcp_for_copilot/mcp_client.py

def __init__(
    self,
    *,
    url: str,
    transport: str = "stdio",
    auth_token: str = "",
    timeout: float = 60.0,
    env: dict[str, str] | None = None,
    transport_impl: httpx.AsyncBaseTransport | None = None,
) -> None:
    normalised = (transport or "").strip().lower()
    if normalised not in SUPPORTED_TRANSPORTS:
        raise MCPClientError(
            f"Unsupported MCP transport {redact(normalised)!r}. "
            f"Supported: {', '.join(SUPPORTED_TRANSPORTS)}"
        )
    if not (url or "").strip():
        raise MCPClientError("MCP server url/command is required")

    self.url = url.strip()
    self.transport = normalised
    self.auth_token = (auth_token or "").strip()
    self.timeout = float(timeout)
    self.env = env
    self._transport_impl = transport_impl

    self._process: asyncio.subprocess.Process | None = None
    self._http: httpx.AsyncClient | None = None
    self._next_id = 0
    self._lock = asyncio.Lock()
    self._initialised = False
    self._server_info: dict[str, Any] = {}
    self._stderr_task: asyncio.Task[None] | None = None
    self._stderr_tail: list[str] = []

call_tool async

call_tool(
    name: str, arguments: dict[str, Any] | None = None
) -> MCPCallResult

Invoke name with arguments and flatten the content blocks.

Source code in src/mcp_for_copilot/mcp_client.py

async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> MCPCallResult:
    """Invoke *name* with *arguments* and flatten the content blocks."""
    if not name:
        raise MCPClientError("Tool name is required")
    result = await self._request("tools/call", {"name": name, "arguments": arguments or {}})
    payload = result or {}
    return MCPCallResult(
        tool=name,
        content=_flatten_content(payload.get("content")),
        is_error=bool(payload.get("isError")),
        raw=payload,
    )

connect async

connect() -> None

Open the transport and perform the MCP initialize handshake.

Source code in src/mcp_for_copilot/mcp_client.py

async def connect(self) -> None:
    """Open the transport and perform the MCP ``initialize`` handshake."""
    if self.transport == "stdio":
        await self._spawn()
    else:
        headers = {"Accept": "application/json, text/event-stream"}
        if self.auth_token:
            headers["Authorization"] = f"Bearer {self.auth_token}"
        self._http = httpx.AsyncClient(
            timeout=self.timeout,
            headers=headers,
            transport=self._transport_impl,
        )

    try:
        result = await self._request(
            "initialize",
            {
                "protocolVersion": _MCP_PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "mcp-for-copilot", "version": "0.1.0"},
            },
        )
        self._server_info = result or {}
        await self._notify("notifications/initialized", {})
        self._initialised = True
        log_info(
            "mcp connected transport=%s server=%s",
            self.transport,
            redact(str(self._server_info.get("serverInfo", {}).get("name", "?"))),
        )
    except Exception:
        await self.close()
        raise

list_tools async

list_tools() -> list[MCPTool]

Return every tool the server advertises.

Source code in src/mcp_for_copilot/mcp_client.py

async def list_tools(self) -> list[MCPTool]:
    """Return every tool the server advertises."""
    result = await self._request("tools/list", {})
    raw_tools = (result or {}).get("tools") or []
    tools = [MCPTool.from_raw(t) for t in raw_tools if isinstance(t, dict) and t.get("name")]
    log_debug("mcp tools/list returned %d tools", len(tools))
    return tools

MCPClientError

Bases: RuntimeError

Raised for any MCP transport or protocol failure.

MCPClientManager

MCPClientManager(
    clients: dict[str, MCPClient] | None = None,
)

Manager for multiple MCP clients.

Source code in src/mcp_for_copilot/mcp_client.py

def __init__(self, clients: dict[str, MCPClient] | None = None):
    self._clients = clients or {}
    self._tool_to_client: dict[str, str] = {}  # Maps tool name to client name

call_tool async

call_tool(
    client_name: str,
    tool_name: str,
    arguments: dict[str, Any] | None = None,
) -> MCPCallResult

Call a tool on a specific client.

Source code in src/mcp_for_copilot/mcp_client.py

async def call_tool(
    self, client_name: str, tool_name: str, arguments: dict[str, Any] | None = None
) -> MCPCallResult:
    """Call a tool on a specific client."""
    client = self._clients.get(client_name)
    if client is None:
        raise MCPClientError(f"No MCP client named {client_name!r}")
    return await client.call_tool(tool_name, arguments)

close async

close() -> None

Close all clients.

Source code in src/mcp_for_copilot/mcp_client.py

async def close(self) -> None:
    """Close all clients."""
    for client in self._clients.values():
        await client.close()

from_settings classmethod

from_settings(settings: Settings) -> MCPClientManager

Create an MCPClientManager from settings.

This reads the MCP servers configuration from settings.mcp_servers_parsed and also handles backward compatibility with the deprecated single server fields.

Source code in src/mcp_for_copilot/mcp_client.py

@classmethod
def from_settings(cls, settings: Settings) -> MCPClientManager:
    """Create an MCPClientManager from settings.

    This reads the MCP servers configuration from settings.mcp_servers_parsed
    and also handles backward compatibility with the deprecated single server fields.
    """
    clients = {}
    # Handle new way: mcp_servers
    for server_config in settings.mcp_servers_parsed:
        name = server_config.get("name")
        if not name:
            continue
        transport = server_config.get("transport", "stdio")
        url_or_command = server_config.get("url") or server_config.get("command")
        auth_token = server_config.get("auth_token", "")
        if not url_or_command:
            continue
        client = MCPClient(
            url=url_or_command,
            transport=transport,
            auth_token=auth_token,
            timeout=settings.mcp_timeout_seconds,
        )
        clients[name] = client

    # Handle backward compatibility: if the deprecated fields are set and we haven't already added a client for that
    if settings.mcp_configured and not clients:
        # We assume the deprecated fields are for a server named "default"
        clients["default"] = MCPClient(
            url=settings.mcp_server_url,
            transport=settings.mcp_transport,
            auth_token=settings.mcp_auth_token,
            timeout=settings.mcp_timeout_seconds,
        )

    return cls(clients)

get_client

get_client(name: str | None = None) -> MCPClient | None

Get a client by name. If name is None, return the first client (for backward compatibility).

Source code in src/mcp_for_copilot/mcp_client.py

def get_client(self, name: str | None = None) -> MCPClient | None:
    """Get a client by name. If name is None, return the first client (for backward compatibility)."""
    if name is None:
        if self._clients:
            return next(iter(self._clients.values()))
        return None
    return self._clients.get(name)

get_client_for_tool

get_client_for_tool(
    tool_name: str,
) -> tuple[str, MCPClient] | None

Get the client that provides the given tool, if any.

Source code in src/mcp_for_copilot/mcp_client.py

def get_client_for_tool(self, tool_name: str) -> tuple[str, MCPClient] | None:
    """Get the client that provides the given tool, if any."""
    client_name = self._tool_to_client.get(tool_name)
    if client_name is None:
        return None
    client = self._clients.get(client_name)
    if client is None:
        # This should not happen if the map is maintained correctly
        return None
    return client_name, client

list_all_tools async

list_all_tools() -> dict[str, list[MCPTool]]

List tools from all clients and update the tool-to-client mapping.

Returns a dictionary mapping client name to list of tools. Also updates the internal _tool_to_client mapping.

Source code in src/mcp_for_copilot/mcp_client.py

async def list_all_tools(self) -> dict[str, list[MCPTool]]:
    """List tools from all clients and update the tool-to-client mapping.

    Returns a dictionary mapping client name to list of tools.
    Also updates the internal _tool_to_client mapping.
    """
    # Clear the existing mapping
    self._tool_to_client.clear()
    result: dict[str, list[MCPTool]] = {}
    for name, client in self._clients.items():
        try:
            tools = await client.list_tools()
            result[name] = tools
            # Update the mapping: for each tool, map to this client
            # If the same tool appears in multiple clients, the last one wins.
            for tool in tools:
                self._tool_to_client[tool.name] = name
        except Exception as exc:
            log_warning("Failed to list tools for MCP client %s: %s", name, exc)
            result[name] = []
    return result

MCPTool dataclass

MCPTool(
    name: str,
    description: str = "",
    input_schema: dict[str, Any] = dict(),
)

A tool advertised by the MCP server.

OrchestratorResult dataclass

OrchestratorResult(
    content: str = "",
    model: str = "",
    rounds: int = 0,
    invocations: list[ToolInvocation] = list(),
    usage: dict[str, Any] = dict(),
    error: str = "",
    plan: list[PlanStep] = list(),
    loop_stopped: bool = False,
)

The final answer plus the audit trail of every tool call.

plan_progress property

plan_progress: dict[str, int]

Count of steps per status, for a progress display.

ProviderResponse dataclass

ProviderResponse(
    content: str = "",
    tool_calls: list[ToolCall] = list(),
    model: str = "",
    finish_reason: str = "",
    usage: dict[str, Any] = dict(),
)

Normalised provider reply.

RoutedTool dataclass

RoutedTool(
    name: str,
    description: str = "",
    input_schema: dict[str, Any] = dict(),
    risky: bool = False,
)

A tool that passed the allowlist, with its declared schema.

RouterDecision dataclass

RouterDecision(
    allowed: bool,
    tool: str,
    reason: str = "",
    needs_approval: bool = False,
    arguments: dict[str, Any] = dict(),
)

The router's verdict for one requested tool call.

Settings

Bases: BaseSettings

Runtime configuration, read from the environment or a .env file.

llm_fallback_configured property

llm_fallback_configured: bool

True when the Z.ai backup is usable (enabled + key + model).

mcp_allowed_tools property

mcp_allowed_tools: tuple[str, ...]

Parsed allowlist. Empty tuple means 'use the default set'.

mcp_servers_parsed property

mcp_servers_parsed: list[dict[str, Any]]

Parsed MCP servers configuration. Returns empty list if not configured or invalid.

describe

describe() -> dict[str, object]

Non-secret summary, safe to log or return over HTTP.

Source code in src/mcp_for_copilot/config.py

def describe(self) -> dict[str, object]:
    """Non-secret summary, safe to log or return over HTTP."""
    return {
        "llm_provider": self.llm_provider,
        "llm_model_id": self.llm_model_id,
        "llm_base_url": self.llm_base_url,
        "llm_api_key_present": bool(self.llm_api_key),
        "llm_reasoning_effort": self.llm_reasoning_effort or None,
        "llm_fallback_enabled": self.llm_fallback_enabled,
        "llm_fallback_configured": self.llm_fallback_configured,
        "llm_fallback_base_url": self.llm_fallback_base_url,
        "llm_fallback_model_id": self.llm_fallback_model_id,
        "llm_fallback_api_key_present": bool(self.llm_fallback_api_key),
        "mcp_transport": self.mcp_transport,
        "mcp_server_url": redact(self.mcp_server_url),
        "mcp_servers": self._describe_mcp_servers(),
        "mcp_configured": self.mcp_configured,
        "mcp_require_approval": self.mcp_require_approval,
        "mcp_max_tool_rounds": self.mcp_max_tool_rounds,
        "mcp_plan_mode": self.mcp_plan_mode,
        "mcp_loop_detection": self.mcp_loop_detection,
        "mcp_loop_soft_threshold": self.mcp_loop_soft_threshold,
        "mcp_loop_hard_threshold": self.mcp_loop_hard_threshold,
        "gateway_api_key_required": bool(self.gateway_api_key),
    }

ToolCall dataclass

ToolCall(
    id: str, name: str, arguments: dict[str, Any] = dict()
)

A tool invocation requested by the model.

from_openai classmethod

from_openai(raw: dict[str, Any]) -> ToolCall

Parse one entry of choices[0].message.tool_calls.

Malformed JSON arguments are surfaced as an empty dict plus a marker key so the router can reject the call instead of crashing.

Source code in src/mcp_for_copilot/provider.py

@classmethod
def from_openai(cls, raw: dict[str, Any]) -> ToolCall:
    """Parse one entry of ``choices[0].message.tool_calls``.

    Malformed JSON arguments are surfaced as an empty dict plus a marker
    key so the router can reject the call instead of crashing.
    """
    function = raw.get("function") or {}
    raw_args = function.get("arguments")
    parsed: dict[str, Any] = {}
    if isinstance(raw_args, dict):
        parsed = raw_args
    elif isinstance(raw_args, str) and raw_args.strip():
        try:
            decoded = json.loads(raw_args)
            if isinstance(decoded, dict):
                parsed = decoded
            else:
                parsed = {"__malformed_arguments__": raw_args[:200]}
        except json.JSONDecodeError:
            parsed = {"__malformed_arguments__": raw_args[:200]}
    return cls(
        id=str(raw.get("id") or ""),
        name=str(function.get("name") or ""),
        arguments=parsed,
    )

ToolInvocation dataclass

ToolInvocation(
    tool: str,
    arguments: dict[str, Any] = dict(),
    allowed: bool = False,
    approved: bool | None = None,
    executed: bool = False,
    is_error: bool = False,
    result: str = "",
    reason: str = "",
)

A record of one tool call and what happened to it.

ToolRouter

ToolRouter(
    *,
    allowed_tools: tuple[str, ...]
    | list[str]
    | None = None,
    require_approval: bool = True,
)

Allowlist + validation gate in front of the MCP server.

Source code in src/mcp_for_copilot/tool_router.py

def __init__(
    self,
    *,
    allowed_tools: tuple[str, ...] | list[str] | None = None,
    require_approval: bool = True,
) -> None:
    self.allowed_tools: tuple[str, ...] = tuple(
        DEFAULT_ALLOWED_TOOLS if allowed_tools is None else allowed_tools
    )
    self.require_approval = require_approval
    self._registry: dict[str, RoutedTool] = {}

available

available() -> list[RoutedTool]

Registered tools that are on the allowlist.

Source code in src/mcp_for_copilot/tool_router.py

def available(self) -> list[RoutedTool]:
    """Registered tools that are on the allowlist."""
    return [t for t in self._registry.values() if t.name in self.allowed_tools]

needs_approval

needs_approval(name: str) -> bool

True when name is risky and approval is enabled.

Source code in src/mcp_for_copilot/tool_router.py

def needs_approval(self, name: str) -> bool:
    """True when *name* is risky and approval is enabled."""
    return bool(self.require_approval and is_risky_tool(name))

register

register(tool: Any) -> RoutedTool | None

Register an MCP tool descriptor (anything with name).

Returns None and skips the tool when it is a gateway self-loop (gpt6_* / gateway_*): registering those would let the model call back into this same orchestrator.

Source code in src/mcp_for_copilot/tool_router.py

def register(self, tool: Any) -> RoutedTool | None:
    """Register an MCP tool descriptor (anything with ``name``).

    Returns ``None`` and skips the tool when it is a gateway self-loop
    (``gpt6_*`` / ``gateway_*``): registering those would let the model
    call back into this same orchestrator.
    """
    name = str(getattr(tool, "name", "") or "")
    if not name:
        raise ToolRouterError("Cannot register a tool without a name")
    if is_self_loop_tool(name):
        log_warning("tool %s skipped: gateway self-loop tool is never served", redact(name))
        return None
    routed = RoutedTool(
        name=name,
        description=str(getattr(tool, "description", "") or ""),
        input_schema=getattr(tool, "input_schema", None) or {},
        risky=is_risky_tool(name),
    )
    self._registry[name] = routed
    return routed

route

route(
    name: str, arguments: dict[str, Any] | None = None
) -> RouterDecision

Decide whether name may run, and whether it needs approval.

Source code in src/mcp_for_copilot/tool_router.py

def route(self, name: str, arguments: dict[str, Any] | None = None) -> RouterDecision:
    """Decide whether *name* may run, and whether it needs approval."""
    args = arguments or {}
    if not name:
        return RouterDecision(False, "", "tool name is empty")

    if is_self_loop_tool(name):
        log_warning("tool %s denied: gateway self-loop tool is never served", redact(name))
        return RouterDecision(
            False,
            name,
            f"tool {name!r} calls back into this gateway and is never served",
            arguments=args,
        )

    if name not in self.allowed_tools:
        log_warning("tool %s denied: not in allowlist", redact(name))
        return RouterDecision(
            False,
            name,
            f"tool {name!r} is not in the allowlist ({len(self.allowed_tools)} tools enabled)",
            arguments=args,
        )

    tool = self._registry.get(name)
    if tool is None:
        # Allowed by name but never advertised by the server: still deny,
        # because we cannot validate its arguments.
        return RouterDecision(
            False,
            name,
            f"tool {name!r} was not advertised by the MCP server",
            arguments=args,
        )

    problems = validate_arguments(tool.input_schema, args)
    if problems:
        return RouterDecision(
            False, name, "invalid arguments: " + "; ".join(problems), arguments=args
        )

    risky = tool.risky
    needs_approval = bool(risky and self.require_approval)
    log_debug("tool %s routed risky=%s approval=%s", redact(name), risky, needs_approval)
    return RouterDecision(
        True,
        name,
        "risky tool" if risky else "read-only tool",
        needs_approval=needs_approval,
        arguments=args,
    )

ToolRouterError

Bases: RuntimeError

Raised when a tool call is rejected.

allow_all_approver

allow_all_approver() -> Approver

Approve every request. Tests and trusted local runs only.

Source code in src/mcp_for_copilot/approval.py

def allow_all_approver() -> Approver:
    """Approve every request. Tests and trusted local runs only."""

    async def _approve(request: ApprovalRequest) -> ApprovalDecision:
        return ApprovalDecision.allow("allow_all approver", approver="allow_all")

    return _approve

contains_secret

contains_secret(text: str) -> bool

True when text still contains something the redactor would remove.

Used by the test-suite to assert that no secret reaches a log sink.

Source code in src/mcp_for_copilot/logging_utils.py

def contains_secret(text: str) -> bool:
    """True when *text* still contains something the redactor would remove.

    Used by the test-suite to assert that no secret reaches a log sink.
    """
    if not text:
        return False
    for pattern, _ in _REDACTION_RULES:
        for match in pattern.finditer(text):
            candidate = match.group(0)
            # Ignore matches that are already redaction markers or placeholders.
            if "REDACTED" in candidate:
                continue
            if _PLACEHOLDER_RE.match(candidate):
                continue
            return True
    return False

deny_all_approver

deny_all_approver() -> Approver

Deny every request.

Source code in src/mcp_for_copilot/approval.py

def deny_all_approver() -> Approver:
    """Deny every request."""

    async def _deny(request: ApprovalRequest) -> ApprovalDecision:
        return ApprovalDecision.deny("deny_all approver", approver="deny_all")

    return _deny

get_settings cached

get_settings() -> Settings

Return the process-wide settings singleton.

Source code in src/mcp_for_copilot/config.py

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """Return the process-wide settings singleton."""
    for candidate in _env_file_candidates():
        _load_dotenv(candidate)
    return Settings()

is_risky_tool

is_risky_tool(name: str) -> bool

Return True when name looks like a mutating operation.

Fails closed: an empty or unparseable name is treated as risky.

Source code in src/mcp_for_copilot/tool_router.py

def is_risky_tool(name: str) -> bool:
    """Return True when *name* looks like a mutating operation.

    Fails closed: an empty or unparseable name is treated as risky.
    """
    if not name:
        return True
    lowered = name.strip().lower()
    if lowered in EXPLICIT_SAFE_TOOLS:
        return False
    if lowered in EXPLICIT_RISKY_TOOLS:
        return True
    words = set(re.split(r"[^a-z0-9]+", lowered))
    return any(pattern in words for pattern in RISKY_TOOL_PATTERNS)

redact

redact(value: Any) -> str

Return value as a string with every credential-looking substring removed.

Safe to call on any object; None becomes "".

Source code in src/mcp_for_copilot/logging_utils.py

def redact(value: Any) -> str:
    """Return *value* as a string with every credential-looking substring removed.

    Safe to call on any object; ``None`` becomes ``""``.
    """
    if value is None:
        return ""
    text = value if isinstance(value, str) else str(value)
    for pattern, replacement in _REDACTION_RULES:
        text = pattern.sub(replacement, text)
    return text

resolve_model_id

resolve_model_id(model_id: str | None = None) -> str

Resolve model_id (or the configured default) to a canonical id.

Raises :class:LLMProviderError for an unknown id so a typo fails loudly instead of silently hitting the provider with a bad name.

Source code in src/mcp_for_copilot/provider.py

def resolve_model_id(model_id: str | None = None) -> str:
    """Resolve *model_id* (or the configured default) to a canonical id.

    Raises :class:`LLMProviderError` for an unknown id so a typo fails loudly
    instead of silently hitting the provider with a bad name.
    """
    candidate = (model_id or get_settings().llm_model_id or "").strip()
    if not candidate:
        raise LLMProviderError("No model configured. Set LLM_MODEL_ID in the environment.")
    canonical = MODEL_ALIASES.get(candidate.lower(), candidate)
    if canonical not in KNOWN_MODELS:
        known = ", ".join(sorted(KNOWN_MODELS))
        raise LLMProviderError(f"Unknown model id {redact(candidate)!r}. Known ids: {known}")
    return canonical

safe_log

safe_log(
    level: int, message: str, *args: Any, **kwargs: Any
) -> None

Log message after redaction.

%-style args are redacted individually so a secret passed as an argument cannot slip through the format string. Scalar args keep their type so %d/%f specifiers still work.

Source code in src/mcp_for_copilot/logging_utils.py

def safe_log(level: int, message: str, *args: Any, **kwargs: Any) -> None:
    """Log *message* after redaction.

    ``%``-style args are redacted individually so a secret passed as an
    argument cannot slip through the format string. Scalar args keep their
    type so ``%d``/``%f`` specifiers still work.
    """
    safe_args = tuple(_redact_arg(a) for a in args)
    logger.log(level, redact(message), *safe_args, **kwargs)

static_approver

static_approver(
    *,
    allow: tuple[str, ...] | list[str] = (),
    deny: tuple[str, ...] | list[str] = (),
) -> Approver

Approve tools in allow, deny tools in deny, deny anything else.

Source code in src/mcp_for_copilot/approval.py

def static_approver(
    *, allow: tuple[str, ...] | list[str] = (), deny: tuple[str, ...] | list[str] = ()
) -> Approver:
    """Approve tools in *allow*, deny tools in *deny*, deny anything else."""
    allowed = set(allow)
    denied = set(deny)

    async def _decide(request: ApprovalRequest) -> ApprovalDecision:
        if request.tool in allowed:
            return ApprovalDecision.allow("static allowlist", approver="static")
        if request.tool in denied:
            return ApprovalDecision.deny("static denylist", approver="static")
        return ApprovalDecision.deny("tool not in the static allowlist", approver="static")

    return _decide

validate_arguments

validate_arguments(
    schema: dict[str, Any], arguments: dict[str, Any]
) -> list[str]

Validate arguments against a JSON-schema subset.

Returns a list of human-readable problems; an empty list means valid. Supports type, required, properties, enum and items. Unknown keywords are ignored rather than rejected, so a richer server schema still works.

Source code in src/mcp_for_copilot/tool_router.py

def validate_arguments(schema: dict[str, Any], arguments: dict[str, Any]) -> list[str]:
    """Validate *arguments* against a JSON-schema subset.

    Returns a list of human-readable problems; an empty list means valid.
    Supports ``type``, ``required``, ``properties``, ``enum`` and ``items``.
    Unknown keywords are ignored rather than rejected, so a richer server
    schema still works.
    """
    problems: list[str] = []
    if not isinstance(schema, dict) or not schema:
        return problems
    if not isinstance(arguments, dict):
        return ["arguments must be a JSON object"]

    expected = schema.get("type")
    if expected and expected != "object":
        problems.append(f"schema declares type {expected!r}, expected 'object'")

    for key in schema.get("required") or []:
        if key not in arguments:
            problems.append(f"missing required argument {key!r}")

    properties = schema.get("properties") or {}
    for key, value in arguments.items():
        if key not in properties:
            if schema.get("additionalProperties") is False:
                problems.append(f"unexpected argument {key!r}")
            continue
        problems.extend(_validate_value(key, value, properties[key] or {}))

    return problems

Configuration

mcp_for_copilot.config

Configuration — env-driven, no hardcoded secrets.

Every value has a safe default so the package imports cleanly with an empty environment. A credential is never given a default: an unset key stays empty and the layer that needs it fails loudly instead of silently using a literal.

Settings

Bases: BaseSettings

Runtime configuration, read from the environment or a .env file.

llm_fallback_configured property

llm_fallback_configured: bool

True when the Z.ai backup is usable (enabled + key + model).

mcp_allowed_tools property

mcp_allowed_tools: tuple[str, ...]

Parsed allowlist. Empty tuple means 'use the default set'.

mcp_servers_parsed property

mcp_servers_parsed: list[dict[str, Any]]

Parsed MCP servers configuration. Returns empty list if not configured or invalid.

describe

describe() -> dict[str, object]

Non-secret summary, safe to log or return over HTTP.

Source code in src/mcp_for_copilot/config.py

def describe(self) -> dict[str, object]:
    """Non-secret summary, safe to log or return over HTTP."""
    return {
        "llm_provider": self.llm_provider,
        "llm_model_id": self.llm_model_id,
        "llm_base_url": self.llm_base_url,
        "llm_api_key_present": bool(self.llm_api_key),
        "llm_reasoning_effort": self.llm_reasoning_effort or None,
        "llm_fallback_enabled": self.llm_fallback_enabled,
        "llm_fallback_configured": self.llm_fallback_configured,
        "llm_fallback_base_url": self.llm_fallback_base_url,
        "llm_fallback_model_id": self.llm_fallback_model_id,
        "llm_fallback_api_key_present": bool(self.llm_fallback_api_key),
        "mcp_transport": self.mcp_transport,
        "mcp_server_url": redact(self.mcp_server_url),
        "mcp_servers": self._describe_mcp_servers(),
        "mcp_configured": self.mcp_configured,
        "mcp_require_approval": self.mcp_require_approval,
        "mcp_max_tool_rounds": self.mcp_max_tool_rounds,
        "mcp_plan_mode": self.mcp_plan_mode,
        "mcp_loop_detection": self.mcp_loop_detection,
        "mcp_loop_soft_threshold": self.mcp_loop_soft_threshold,
        "mcp_loop_hard_threshold": self.mcp_loop_hard_threshold,
        "gateway_api_key_required": bool(self.gateway_api_key),
    }

get_settings cached

get_settings() -> Settings

Return the process-wide settings singleton.

Source code in src/mcp_for_copilot/config.py

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """Return the process-wide settings singleton."""
    for candidate in _env_file_candidates():
        _load_dotenv(candidate)
    return Settings()

reset_settings_cache

reset_settings_cache() -> None

Clear the settings cache. Intended for tests.

Source code in src/mcp_for_copilot/config.py

def reset_settings_cache() -> None:
    """Clear the settings cache. Intended for tests."""
    get_settings.cache_clear()

Provider

mcp_for_copilot.provider

LLM provider adapter.

The single place in the package that talks to a language model. Every other layer (router, approval, orchestrator) is provider-agnostic and only sees the dataclasses defined here.

Design rules

• No hardcoded model id. The model always comes from settings.llm_model_id. A caller may override it per request, but the default is never a literal inside this module.
• No hardcoded credential. The key comes from settings.llm_api_key.
• No secret in an error. Every raised message is passed through :func:mcp_for_copilot.logging_utils.redact.
• Tools are passed through, never invented. The adapter forwards the tool list it is given; it does not add tools of its own.

LLMProvider

LLMProvider(
    *,
    provider: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    model_id: str | None = None,
    timeout: float | None = None,
    transport: AsyncBaseTransport | None = None,
    fallback_api_key: str | None = None,
    fallback_base_url: str | None = None,
    fallback_model_id: str | None = None,
    fallback_enabled: bool | None = None,
)

OpenAI-compatible chat-completions adapter.

Only openai is implemented. An unknown LLM_PROVIDER raises rather than falling back, so a misconfiguration is never masked.

Source code in src/mcp_for_copilot/provider.py

def __init__(
    self,
    *,
    provider: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    model_id: str | None = None,
    timeout: float | None = None,
    transport: httpx.AsyncBaseTransport | None = None,
    fallback_api_key: str | None = None,
    fallback_base_url: str | None = None,
    fallback_model_id: str | None = None,
    fallback_enabled: bool | None = None,
) -> None:
    settings = get_settings()
    self.provider = (provider or settings.llm_provider or "").strip().lower()
    if self.provider not in SUPPORTED_PROVIDERS:
        raise LLMProviderError(
            f"Unsupported LLM_PROVIDER {redact(self.provider)!r}. "
            f"Supported: {', '.join(SUPPORTED_PROVIDERS)}"
        )
    self.api_key = (api_key if api_key is not None else settings.llm_api_key).strip()
    self.base_url = (
        (base_url if base_url is not None else settings.llm_base_url).strip().rstrip("/")
    )
    self.model_id = resolve_model_id(model_id or settings.llm_model_id or None)
    self.timeout = float(timeout if timeout is not None else settings.llm_timeout_seconds)
    # Injectable transport so tests never touch the network.
    self._transport = transport

    # -- Z.ai (GLM) backup provider -----------------------------------
    # Used only when the primary request fails with a transport error,
    # timeout, or 5xx. A 4xx is a caller/request problem, not an outage,
    # so it is never retried against the fallback.
    self.fallback_enabled = (
        settings.llm_fallback_enabled if fallback_enabled is None else fallback_enabled
    )
    self.fallback_api_key = (
        fallback_api_key if fallback_api_key is not None else settings.llm_fallback_api_key
    ).strip()
    self.fallback_base_url = (
        fallback_base_url
        if fallback_base_url is not None
        else settings.llm_fallback_base_url
    ).strip().rstrip("/")
    self.fallback_model_id = (
        fallback_model_id
        if fallback_model_id is not None
        else settings.llm_fallback_model_id
    ).strip()
    self.fallback_timeout = float(settings.llm_fallback_timeout_seconds)

fallback_configured property

fallback_configured: bool

True when the Z.ai backup is usable (enabled + key + model).

complete async

complete(
    messages: list[dict[str, Any]],
    *,
    tools: list[dict[str, Any]] | None = None,
    model_id: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    reasoning_effort: str | None = None,
) -> ProviderResponse

Send messages to the provider and return a normalised reply.

tools must already be in OpenAI function-calling format; the adapter does not transform or extend it.

When the primary provider fails with a transport error, timeout, or 5xx and a Z.ai fallback is configured, the same request is retried once against the fallback endpoint. A 4xx is never retried: it is a request problem, not an outage.

Source code in src/mcp_for_copilot/provider.py

async def complete(
    self,
    messages: list[dict[str, Any]],
    *,
    tools: list[dict[str, Any]] | None = None,
    model_id: str | None = None,
    max_tokens: int | None = None,
    temperature: float | None = None,
    reasoning_effort: str | None = None,
) -> ProviderResponse:
    """Send *messages* to the provider and return a normalised reply.

    ``tools`` must already be in OpenAI function-calling format; the
    adapter does not transform or extend it.

    When the primary provider fails with a transport error, timeout, or
    5xx and a Z.ai fallback is configured, the same request is retried
    once against the fallback endpoint. A 4xx is never retried: it is a
    request problem, not an outage.
    """
    if not self.api_key:
        raise LLMProviderError(
            "LLM_API_KEY is not configured. Set it in the environment (never in source)."
        )

    model = resolve_model_id(model_id or self.model_id)

    # A client-facing tier id (gpt-6-luna / gpt-6-sol / gpt-6-astra) is not
    # a real DeepSeek model. Translate it to deepseek-flash + the tier's
    # reasoning_effort before anything is sent upstream, otherwise DeepSeek
    # rejects the request with HTTP 400.
    #
    # This translation is ONLY valid when the upstream actually IS
    # DeepSeek. gpt-6-luna / gpt-6-sol / gpt-6-astra are also real OpenAI
    # model ids, so applying the table unconditionally rewrites them to
    # `deepseek-flash` and sends that to OpenAI, which answers
    #   404 "The model `deepseek-flash` does not exist"
    # and the gateway surfaces a 502. Gate on the provider (and on the
    # base_url as a belt-and-braces check for a DeepSeek-compatible host).
    if self._uses_deepseek():
        tier = resolve_deepseek_tier(model)
        if tier is not None:
            model = tier["model"]
            if reasoning_effort is None:
                reasoning_effort = tier["reasoning_effort"]
            if max_tokens is None:
                max_tokens = tier.get("max_tokens")

    body = self._build_body(
        model,
        messages,
        tools=tools,
        max_tokens=max_tokens,
        temperature=temperature,
        reasoning_effort=reasoning_effort,
    )

    # Route glm-* to Z.ai when the primary upstream is not Z.ai -- the
    # primary would 404 on every glm id and 4xx never hits the fallback.
    glm_route = self._glm_route() if model.startswith("glm-") else None
    if glm_route is not None:
        route_url, route_key, route_timeout = glm_route
        log_debug(
            "llm request provider=zai-primary-override model=%s messages=%d tools=%d",
            model,
            len(messages),
            len(tools or []),
        )
        return await self._post(
            base_url=route_url,
            api_key=route_key,
            body=body,
            timeout=route_timeout,
        )

    log_debug(
        "llm request provider=%s model=%s messages=%d tools=%d",
        self.provider,
        model,
        len(messages),
        len(tools or []),
    )

    try:
        return await self._post(
            base_url=self.base_url,
            api_key=self.api_key,
            body=body,
            timeout=self.timeout,
        )
    except LLMProviderError as primary_error:
        if not self.fallback_configured or not self._is_retryable(primary_error):
            raise
        log_warning(
            "primary LLM failed (%s); retrying on Z.ai fallback model=%s",
            redact(str(primary_error)),
            self.fallback_model_id,
        )
        return await self._complete_fallback(
            messages,
            tools=tools,
            max_tokens=max_tokens,
            temperature=temperature,
            reasoning_effort=reasoning_effort,
            primary_error=primary_error,
        )

describe

describe() -> dict[str, Any]

Non-secret description of the adapter, safe to return over HTTP.

Source code in src/mcp_for_copilot/provider.py

def describe(self) -> dict[str, Any]:
    """Non-secret description of the adapter, safe to return over HTTP."""
    return {
        "provider": self.provider,
        "model_id": self.model_id,
        "base_url": self.base_url,
        "api_key_present": bool(self.api_key),
        "timeout_seconds": self.timeout,
        "fallback_enabled": self.fallback_enabled,
        "fallback_configured": self.fallback_configured,
        "fallback_base_url": self.fallback_base_url,
        "fallback_model_id": self.fallback_model_id,
        "fallback_api_key_present": bool(self.fallback_api_key),
        "known_models": sorted(KNOWN_MODELS),
    }

to_openai_tools staticmethod

to_openai_tools(tools: list[Any]) -> list[dict[str, Any]]

Convert MCP tool descriptors into OpenAI function-calling schemas.

Accepts objects exposing name / description / input_schema (i.e. :class:mcp_for_copilot.mcp_client.MCPTool).

Source code in src/mcp_for_copilot/provider.py

@staticmethod
def to_openai_tools(tools: list[Any]) -> list[dict[str, Any]]:
    """Convert MCP tool descriptors into OpenAI function-calling schemas.

    Accepts objects exposing ``name`` / ``description`` / ``input_schema``
    (i.e. :class:`mcp_for_copilot.mcp_client.MCPTool`).
    """
    converted: list[dict[str, Any]] = []
    for tool in tools:
        name = getattr(tool, "name", None)
        if not name:
            continue
        schema = getattr(tool, "input_schema", None) or {
            "type": "object",
            "properties": {},
        }
        converted.append(
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": getattr(tool, "description", "") or "",
                    "parameters": schema,
                },
            }
        )
    return converted

LLMProviderError

Bases: RuntimeError

Raised for any provider failure. The message is always redacted.

ProviderResponse dataclass

ProviderResponse(
    content: str = "",
    tool_calls: list[ToolCall] = list(),
    model: str = "",
    finish_reason: str = "",
    usage: dict[str, Any] = dict(),
)

Normalised provider reply.

ToolCall dataclass

ToolCall(
    id: str, name: str, arguments: dict[str, Any] = dict()
)

A tool invocation requested by the model.

from_openai classmethod

from_openai(raw: dict[str, Any]) -> ToolCall

Parse one entry of choices[0].message.tool_calls.

Malformed JSON arguments are surfaced as an empty dict plus a marker key so the router can reject the call instead of crashing.

Source code in src/mcp_for_copilot/provider.py

@classmethod
def from_openai(cls, raw: dict[str, Any]) -> ToolCall:
    """Parse one entry of ``choices[0].message.tool_calls``.

    Malformed JSON arguments are surfaced as an empty dict plus a marker
    key so the router can reject the call instead of crashing.
    """
    function = raw.get("function") or {}
    raw_args = function.get("arguments")
    parsed: dict[str, Any] = {}
    if isinstance(raw_args, dict):
        parsed = raw_args
    elif isinstance(raw_args, str) and raw_args.strip():
        try:
            decoded = json.loads(raw_args)
            if isinstance(decoded, dict):
                parsed = decoded
            else:
                parsed = {"__malformed_arguments__": raw_args[:200]}
        except json.JSONDecodeError:
            parsed = {"__malformed_arguments__": raw_args[:200]}
    return cls(
        id=str(raw.get("id") or ""),
        name=str(function.get("name") or ""),
        arguments=parsed,
    )

aliases_for

aliases_for(model_id: str) -> list[str]

Every alias that resolves to model_id (sorted).

Source code in src/mcp_for_copilot/provider.py

def aliases_for(model_id: str) -> list[str]:
    """Every alias that resolves to *model_id* (sorted)."""
    canonical = MODEL_ALIASES.get(model_id.lower(), model_id)
    return sorted(a for a, target in MODEL_ALIASES.items() if target == canonical)

log_provider_summary

log_provider_summary(provider: LLMProvider) -> None

Emit a one-line, secret-free summary of the adapter configuration.

Source code in src/mcp_for_copilot/provider.py

def log_provider_summary(provider: LLMProvider) -> None:
    """Emit a one-line, secret-free summary of the adapter configuration."""
    info = provider.describe()
    log_info(
        "llm provider ready provider=%s model=%s key_present=%s",
        info["provider"],
        info["model_id"],
        info["api_key_present"],
    )

resolve_deepseek_tier

resolve_deepseek_tier(
    model_id: str | None,
) -> dict[str, Any] | None

Translate a client-facing tier id to its DeepSeek model + effort.

Returns None when model_id is not a DeepSeek tier, so callers can fall through to the normal alias resolution.

Source code in src/mcp_for_copilot/provider.py

def resolve_deepseek_tier(model_id: str | None) -> dict[str, Any] | None:
    """Translate a client-facing tier id to its DeepSeek model + effort.

    Returns ``None`` when *model_id* is not a DeepSeek tier, so callers can
    fall through to the normal alias resolution.
    """
    if not model_id:
        return None
    candidate = model_id.strip().lower()
    canonical = MODEL_ALIASES.get(candidate, candidate)
    return DEEPSEEK_TIERS.get(canonical)

resolve_model_id

resolve_model_id(model_id: str | None = None) -> str

Resolve model_id (or the configured default) to a canonical id.

Raises :class:LLMProviderError for an unknown id so a typo fails loudly instead of silently hitting the provider with a bad name.

Source code in src/mcp_for_copilot/provider.py

def resolve_model_id(model_id: str | None = None) -> str:
    """Resolve *model_id* (or the configured default) to a canonical id.

    Raises :class:`LLMProviderError` for an unknown id so a typo fails loudly
    instead of silently hitting the provider with a bad name.
    """
    candidate = (model_id or get_settings().llm_model_id or "").strip()
    if not candidate:
        raise LLMProviderError("No model configured. Set LLM_MODEL_ID in the environment.")
    canonical = MODEL_ALIASES.get(candidate.lower(), candidate)
    if canonical not in KNOWN_MODELS:
        known = ", ".join(sorted(KNOWN_MODELS))
        raise LLMProviderError(f"Unknown model id {redact(candidate)!r}. Known ids: {known}")
    return canonical

MCP client

mcp_for_copilot.mcp_client

MCP client — talks to a Model Context Protocol server.

Supports three transports:

stdio Spawns a subprocess and speaks newline-delimited JSON-RPC over its stdin/stdout. This is the transport used by mcp-for-copilot mcp. http Streamable HTTP: one POST per JSON-RPC message. sse Server-Sent Events: POST to the message endpoint, read the reply stream.

The client is an async context manager::

async with MCPClient(url="python -m my_server", transport="stdio") as client:
    tools = await client.list_tools()
    result = await client.call_tool("read_file", {"path": "README.md"})

MCPCallResult dataclass

MCPCallResult(
    tool: str,
    content: str = "",
    is_error: bool = False,
    raw: dict[str, Any] = dict(),
)

The outcome of one tools/call.

MCPClient

MCPClient(
    *,
    url: str,
    transport: str = "stdio",
    auth_token: str = "",
    timeout: float = 60.0,
    env: dict[str, str] | None = None,
    transport_impl: AsyncBaseTransport | None = None,
)

Async MCP client for stdio, http and sse transports.

Source code in src/mcp_for_copilot/mcp_client.py

def __init__(
    self,
    *,
    url: str,
    transport: str = "stdio",
    auth_token: str = "",
    timeout: float = 60.0,
    env: dict[str, str] | None = None,
    transport_impl: httpx.AsyncBaseTransport | None = None,
) -> None:
    normalised = (transport or "").strip().lower()
    if normalised not in SUPPORTED_TRANSPORTS:
        raise MCPClientError(
            f"Unsupported MCP transport {redact(normalised)!r}. "
            f"Supported: {', '.join(SUPPORTED_TRANSPORTS)}"
        )
    if not (url or "").strip():
        raise MCPClientError("MCP server url/command is required")

    self.url = url.strip()
    self.transport = normalised
    self.auth_token = (auth_token or "").strip()
    self.timeout = float(timeout)
    self.env = env
    self._transport_impl = transport_impl

    self._process: asyncio.subprocess.Process | None = None
    self._http: httpx.AsyncClient | None = None
    self._next_id = 0
    self._lock = asyncio.Lock()
    self._initialised = False
    self._server_info: dict[str, Any] = {}
    self._stderr_task: asyncio.Task[None] | None = None
    self._stderr_tail: list[str] = []

call_tool async

call_tool(
    name: str, arguments: dict[str, Any] | None = None
) -> MCPCallResult

Invoke name with arguments and flatten the content blocks.

Source code in src/mcp_for_copilot/mcp_client.py

async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> MCPCallResult:
    """Invoke *name* with *arguments* and flatten the content blocks."""
    if not name:
        raise MCPClientError("Tool name is required")
    result = await self._request("tools/call", {"name": name, "arguments": arguments or {}})
    payload = result or {}
    return MCPCallResult(
        tool=name,
        content=_flatten_content(payload.get("content")),
        is_error=bool(payload.get("isError")),
        raw=payload,
    )

connect async

connect() -> None

Open the transport and perform the MCP initialize handshake.

Source code in src/mcp_for_copilot/mcp_client.py

async def connect(self) -> None:
    """Open the transport and perform the MCP ``initialize`` handshake."""
    if self.transport == "stdio":
        await self._spawn()
    else:
        headers = {"Accept": "application/json, text/event-stream"}
        if self.auth_token:
            headers["Authorization"] = f"Bearer {self.auth_token}"
        self._http = httpx.AsyncClient(
            timeout=self.timeout,
            headers=headers,
            transport=self._transport_impl,
        )

    try:
        result = await self._request(
            "initialize",
            {
                "protocolVersion": _MCP_PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "mcp-for-copilot", "version": "0.1.0"},
            },
        )
        self._server_info = result or {}
        await self._notify("notifications/initialized", {})
        self._initialised = True
        log_info(
            "mcp connected transport=%s server=%s",
            self.transport,
            redact(str(self._server_info.get("serverInfo", {}).get("name", "?"))),
        )
    except Exception:
        await self.close()
        raise

list_tools async

list_tools() -> list[MCPTool]

Return every tool the server advertises.

Source code in src/mcp_for_copilot/mcp_client.py

async def list_tools(self) -> list[MCPTool]:
    """Return every tool the server advertises."""
    result = await self._request("tools/list", {})
    raw_tools = (result or {}).get("tools") or []
    tools = [MCPTool.from_raw(t) for t in raw_tools if isinstance(t, dict) and t.get("name")]
    log_debug("mcp tools/list returned %d tools", len(tools))
    return tools

MCPClientError

Bases: RuntimeError

Raised for any MCP transport or protocol failure.

MCPClientManager

MCPClientManager(
    clients: dict[str, MCPClient] | None = None,
)

Manager for multiple MCP clients.

Source code in src/mcp_for_copilot/mcp_client.py

def __init__(self, clients: dict[str, MCPClient] | None = None):
    self._clients = clients or {}
    self._tool_to_client: dict[str, str] = {}  # Maps tool name to client name

call_tool async

call_tool(
    client_name: str,
    tool_name: str,
    arguments: dict[str, Any] | None = None,
) -> MCPCallResult

Call a tool on a specific client.

Source code in src/mcp_for_copilot/mcp_client.py

async def call_tool(
    self, client_name: str, tool_name: str, arguments: dict[str, Any] | None = None
) -> MCPCallResult:
    """Call a tool on a specific client."""
    client = self._clients.get(client_name)
    if client is None:
        raise MCPClientError(f"No MCP client named {client_name!r}")
    return await client.call_tool(tool_name, arguments)

close async

close() -> None

Close all clients.

Source code in src/mcp_for_copilot/mcp_client.py

async def close(self) -> None:
    """Close all clients."""
    for client in self._clients.values():
        await client.close()

from_settings classmethod

from_settings(settings: Settings) -> MCPClientManager

Create an MCPClientManager from settings.

This reads the MCP servers configuration from settings.mcp_servers_parsed and also handles backward compatibility with the deprecated single server fields.

Source code in src/mcp_for_copilot/mcp_client.py

@classmethod
def from_settings(cls, settings: Settings) -> MCPClientManager:
    """Create an MCPClientManager from settings.

    This reads the MCP servers configuration from settings.mcp_servers_parsed
    and also handles backward compatibility with the deprecated single server fields.
    """
    clients = {}
    # Handle new way: mcp_servers
    for server_config in settings.mcp_servers_parsed:
        name = server_config.get("name")
        if not name:
            continue
        transport = server_config.get("transport", "stdio")
        url_or_command = server_config.get("url") or server_config.get("command")
        auth_token = server_config.get("auth_token", "")
        if not url_or_command:
            continue
        client = MCPClient(
            url=url_or_command,
            transport=transport,
            auth_token=auth_token,
            timeout=settings.mcp_timeout_seconds,
        )
        clients[name] = client

    # Handle backward compatibility: if the deprecated fields are set and we haven't already added a client for that
    if settings.mcp_configured and not clients:
        # We assume the deprecated fields are for a server named "default"
        clients["default"] = MCPClient(
            url=settings.mcp_server_url,
            transport=settings.mcp_transport,
            auth_token=settings.mcp_auth_token,
            timeout=settings.mcp_timeout_seconds,
        )

    return cls(clients)

get_client

get_client(name: str | None = None) -> MCPClient | None

Get a client by name. If name is None, return the first client (for backward compatibility).

Source code in src/mcp_for_copilot/mcp_client.py

def get_client(self, name: str | None = None) -> MCPClient | None:
    """Get a client by name. If name is None, return the first client (for backward compatibility)."""
    if name is None:
        if self._clients:
            return next(iter(self._clients.values()))
        return None
    return self._clients.get(name)

get_client_for_tool

get_client_for_tool(
    tool_name: str,
) -> tuple[str, MCPClient] | None

Get the client that provides the given tool, if any.

Source code in src/mcp_for_copilot/mcp_client.py

def get_client_for_tool(self, tool_name: str) -> tuple[str, MCPClient] | None:
    """Get the client that provides the given tool, if any."""
    client_name = self._tool_to_client.get(tool_name)
    if client_name is None:
        return None
    client = self._clients.get(client_name)
    if client is None:
        # This should not happen if the map is maintained correctly
        return None
    return client_name, client

list_all_tools async

list_all_tools() -> dict[str, list[MCPTool]]

List tools from all clients and update the tool-to-client mapping.

Returns a dictionary mapping client name to list of tools. Also updates the internal _tool_to_client mapping.

Source code in src/mcp_for_copilot/mcp_client.py

async def list_all_tools(self) -> dict[str, list[MCPTool]]:
    """List tools from all clients and update the tool-to-client mapping.

    Returns a dictionary mapping client name to list of tools.
    Also updates the internal _tool_to_client mapping.
    """
    # Clear the existing mapping
    self._tool_to_client.clear()
    result: dict[str, list[MCPTool]] = {}
    for name, client in self._clients.items():
        try:
            tools = await client.list_tools()
            result[name] = tools
            # Update the mapping: for each tool, map to this client
            # If the same tool appears in multiple clients, the last one wins.
            for tool in tools:
                self._tool_to_client[tool.name] = name
        except Exception as exc:
            log_warning("Failed to list tools for MCP client %s: %s", name, exc)
            result[name] = []
    return result

MCPTool dataclass

MCPTool(
    name: str,
    description: str = "",
    input_schema: dict[str, Any] = dict(),
)

A tool advertised by the MCP server.

default_stdio_command

default_stdio_command() -> str

Command used when MCP_SERVER_URL is unset for the stdio transport.

Source code in src/mcp_for_copilot/mcp_client.py

def default_stdio_command() -> str:
    """Command used when ``MCP_SERVER_URL`` is unset for the stdio transport."""
    return f"{shlex.quote(sys.executable)} -m mcp_for_copilot.gateway.server"

split_command

split_command(command: str) -> list[str]

Split a stdio command line into argv.

shlex.split(posix=False) keeps the surrounding quotes on Windows, so "C:\Python\python.exe" -m server would try to execute a file literally named "C:\Python\python.exe". POSIX mode strips the quotes correctly for both platforms, but it also eats backslashes, so Windows paths are handled by a small dedicated scanner instead.

Source code in src/mcp_for_copilot/mcp_client.py

def split_command(command: str) -> list[str]:
    """Split a stdio command line into argv.

    ``shlex.split(posix=False)`` keeps the surrounding quotes on Windows, so
    ``"C:\\Python\\python.exe" -m server`` would try to execute a file literally
    named ``"C:\\Python\\python.exe"``. POSIX mode strips the quotes correctly
    for both platforms, but it also eats backslashes, so Windows paths are
    handled by a small dedicated scanner instead.
    """
    if os.name != "nt":
        return shlex.split(command, posix=True)

    argv: list[str] = []
    current: list[str] = []
    quote: str | None = None
    started = False

    for char in command:
        if quote:
            if char == quote:
                quote = None
            else:
                current.append(char)
            continue
        if char in ("'", '"'):
            quote = char
            started = True
            continue
        if char.isspace():
            if started:
                argv.append("".join(current))
                current = []
                started = False
            continue
        current.append(char)
        started = True

    if started:
        argv.append("".join(current))
    return argv

Tool router

mcp_for_copilot.tool_router

Tool router — the allowlist and argument-validation gate.

Every tool call the model requests passes through here before it can reach the MCP server. The router answers three questions:

1. Is this tool allowed at all? (allowlist)
2. Is it risky? (needs an approval decision)
3. Are the arguments well-formed? (JSON-schema subset validation)

The router fails closed: an unknown tool is denied, and a tool whose name matches a risky verb is treated as risky even if it is not on the explicit list.

RoutedTool dataclass

RoutedTool(
    name: str,
    description: str = "",
    input_schema: dict[str, Any] = dict(),
    risky: bool = False,
)

A tool that passed the allowlist, with its declared schema.

RouterDecision dataclass

RouterDecision(
    allowed: bool,
    tool: str,
    reason: str = "",
    needs_approval: bool = False,
    arguments: dict[str, Any] = dict(),
)

The router's verdict for one requested tool call.

ToolRouter

ToolRouter(
    *,
    allowed_tools: tuple[str, ...]
    | list[str]
    | None = None,
    require_approval: bool = True,
)

Allowlist + validation gate in front of the MCP server.

Source code in src/mcp_for_copilot/tool_router.py

def __init__(
    self,
    *,
    allowed_tools: tuple[str, ...] | list[str] | None = None,
    require_approval: bool = True,
) -> None:
    self.allowed_tools: tuple[str, ...] = tuple(
        DEFAULT_ALLOWED_TOOLS if allowed_tools is None else allowed_tools
    )
    self.require_approval = require_approval
    self._registry: dict[str, RoutedTool] = {}

available

available() -> list[RoutedTool]

Registered tools that are on the allowlist.

Source code in src/mcp_for_copilot/tool_router.py

def available(self) -> list[RoutedTool]:
    """Registered tools that are on the allowlist."""
    return [t for t in self._registry.values() if t.name in self.allowed_tools]

needs_approval

needs_approval(name: str) -> bool

True when name is risky and approval is enabled.

Source code in src/mcp_for_copilot/tool_router.py

def needs_approval(self, name: str) -> bool:
    """True when *name* is risky and approval is enabled."""
    return bool(self.require_approval and is_risky_tool(name))

register

register(tool: Any) -> RoutedTool | None

Register an MCP tool descriptor (anything with name).

Returns None and skips the tool when it is a gateway self-loop (gpt6_* / gateway_*): registering those would let the model call back into this same orchestrator.

Source code in src/mcp_for_copilot/tool_router.py

def register(self, tool: Any) -> RoutedTool | None:
    """Register an MCP tool descriptor (anything with ``name``).

    Returns ``None`` and skips the tool when it is a gateway self-loop
    (``gpt6_*`` / ``gateway_*``): registering those would let the model
    call back into this same orchestrator.
    """
    name = str(getattr(tool, "name", "") or "")
    if not name:
        raise ToolRouterError("Cannot register a tool without a name")
    if is_self_loop_tool(name):
        log_warning("tool %s skipped: gateway self-loop tool is never served", redact(name))
        return None
    routed = RoutedTool(
        name=name,
        description=str(getattr(tool, "description", "") or ""),
        input_schema=getattr(tool, "input_schema", None) or {},
        risky=is_risky_tool(name),
    )
    self._registry[name] = routed
    return routed

route

route(
    name: str, arguments: dict[str, Any] | None = None
) -> RouterDecision

Decide whether name may run, and whether it needs approval.

Source code in src/mcp_for_copilot/tool_router.py

def route(self, name: str, arguments: dict[str, Any] | None = None) -> RouterDecision:
    """Decide whether *name* may run, and whether it needs approval."""
    args = arguments or {}
    if not name:
        return RouterDecision(False, "", "tool name is empty")

    if is_self_loop_tool(name):
        log_warning("tool %s denied: gateway self-loop tool is never served", redact(name))
        return RouterDecision(
            False,
            name,
            f"tool {name!r} calls back into this gateway and is never served",
            arguments=args,
        )

    if name not in self.allowed_tools:
        log_warning("tool %s denied: not in allowlist", redact(name))
        return RouterDecision(
            False,
            name,
            f"tool {name!r} is not in the allowlist ({len(self.allowed_tools)} tools enabled)",
            arguments=args,
        )

    tool = self._registry.get(name)
    if tool is None:
        # Allowed by name but never advertised by the server: still deny,
        # because we cannot validate its arguments.
        return RouterDecision(
            False,
            name,
            f"tool {name!r} was not advertised by the MCP server",
            arguments=args,
        )

    problems = validate_arguments(tool.input_schema, args)
    if problems:
        return RouterDecision(
            False, name, "invalid arguments: " + "; ".join(problems), arguments=args
        )

    risky = tool.risky
    needs_approval = bool(risky and self.require_approval)
    log_debug("tool %s routed risky=%s approval=%s", redact(name), risky, needs_approval)
    return RouterDecision(
        True,
        name,
        "risky tool" if risky else "read-only tool",
        needs_approval=needs_approval,
        arguments=args,
    )

ToolRouterError

Bases: RuntimeError

Raised when a tool call is rejected.

is_risky_tool

is_risky_tool(name: str) -> bool

Return True when name looks like a mutating operation.

Fails closed: an empty or unparseable name is treated as risky.

Source code in src/mcp_for_copilot/tool_router.py

def is_risky_tool(name: str) -> bool:
    """Return True when *name* looks like a mutating operation.

    Fails closed: an empty or unparseable name is treated as risky.
    """
    if not name:
        return True
    lowered = name.strip().lower()
    if lowered in EXPLICIT_SAFE_TOOLS:
        return False
    if lowered in EXPLICIT_RISKY_TOOLS:
        return True
    words = set(re.split(r"[^a-z0-9]+", lowered))
    return any(pattern in words for pattern in RISKY_TOOL_PATTERNS)

is_self_loop_tool

is_self_loop_tool(name: str) -> bool

True when name is a front-end for this same gateway (never servable).

gateway_chat (the HTTP bridge) and the gpt6_* tools (the bundled stdio server) both call back into this orchestrator, so serving them to the model would loop: model -> gateway -> gateway -> model.

Source code in src/mcp_for_copilot/tool_router.py

def is_self_loop_tool(name: str) -> bool:
    """True when *name* is a front-end for this same gateway (never servable).

    ``gateway_chat`` (the HTTP bridge) and the ``gpt6_*`` tools (the bundled
    stdio server) both call back into this orchestrator, so serving them to
    the model would loop: model -> gateway -> gateway -> model.
    """
    lowered = name.lower()
    return any(lowered.startswith(prefix) for prefix in SELF_LOOP_TOOL_PREFIXES)

validate_arguments

validate_arguments(
    schema: dict[str, Any], arguments: dict[str, Any]
) -> list[str]

Validate arguments against a JSON-schema subset.

Returns a list of human-readable problems; an empty list means valid. Supports type, required, properties, enum and items. Unknown keywords are ignored rather than rejected, so a richer server schema still works.

Source code in src/mcp_for_copilot/tool_router.py

def validate_arguments(schema: dict[str, Any], arguments: dict[str, Any]) -> list[str]:
    """Validate *arguments* against a JSON-schema subset.

    Returns a list of human-readable problems; an empty list means valid.
    Supports ``type``, ``required``, ``properties``, ``enum`` and ``items``.
    Unknown keywords are ignored rather than rejected, so a richer server
    schema still works.
    """
    problems: list[str] = []
    if not isinstance(schema, dict) or not schema:
        return problems
    if not isinstance(arguments, dict):
        return ["arguments must be a JSON object"]

    expected = schema.get("type")
    if expected and expected != "object":
        problems.append(f"schema declares type {expected!r}, expected 'object'")

    for key in schema.get("required") or []:
        if key not in arguments:
            problems.append(f"missing required argument {key!r}")

    properties = schema.get("properties") or {}
    for key, value in arguments.items():
        if key not in properties:
            if schema.get("additionalProperties") is False:
                problems.append(f"unexpected argument {key!r}")
            continue
        problems.extend(_validate_value(key, value, properties[key] or {}))

    return problems

Approval

mcp_for_copilot.approval

Approval layer — the human-in-the-loop gate for risky tools.

The orchestrator asks this layer before dispatching any tool the router marked as needing approval. The layer fails closed: if no approver is configured, every request is denied rather than silently allowed.

Three approvers ship with the package:

allow_all_approver Approves everything. For tests and trusted local runs only. deny_all_approver Denies everything. The default when nothing is configured. static_approver Approves or denies a fixed set of tool names.

ApprovalDecision dataclass

ApprovalDecision(
    approved: bool, reason: str = "", approver: str = ""
)

The approver's verdict.

ApprovalLayer

ApprovalLayer(approver: Approver | None = None)

Wraps an approver callable and enforces fail-closed behaviour.

Source code in src/mcp_for_copilot/approval.py

def __init__(self, approver: Approver | None = None) -> None:
    self._approver = approver

request async

request(request: ApprovalRequest) -> ApprovalDecision

Ask the approver. Denies when no approver is configured.

Source code in src/mcp_for_copilot/approval.py

async def request(self, request: ApprovalRequest) -> ApprovalDecision:
    """Ask the approver. Denies when no approver is configured."""
    if self._approver is None:
        log_warning(
            "approval denied for %s: no approver configured (fail closed)",
            redact(request.tool),
        )
        return ApprovalDecision.deny(
            "no approver configured; risky tools are denied by default",
            approver="none",
        )

    try:
        decision = await self._approver(request)
    except Exception as exc:
        log_warning("approver raised for %s: %s", redact(request.tool), type(exc).__name__)
        return ApprovalDecision.deny(f"approver raised {type(exc).__name__}", approver="error")

    if not isinstance(decision, ApprovalDecision):
        return ApprovalDecision.deny(
            "approver returned an invalid decision", approver="invalid"
        )

    log_info(
        "approval %s for %s by %s",
        "granted" if decision.approved else "denied",
        redact(request.tool),
        redact(decision.approver or "unknown"),
    )
    return decision

ApprovalRequest dataclass

ApprovalRequest(
    tool: str,
    arguments: dict[str, Any] = dict(),
    reason: str = "",
    round_index: int = 0,
)

What the orchestrator asks the approver to decide on.

allow_all_approver

allow_all_approver() -> Approver

Approve every request. Tests and trusted local runs only.

Source code in src/mcp_for_copilot/approval.py

def allow_all_approver() -> Approver:
    """Approve every request. Tests and trusted local runs only."""

    async def _approve(request: ApprovalRequest) -> ApprovalDecision:
        return ApprovalDecision.allow("allow_all approver", approver="allow_all")

    return _approve

deny_all_approver

deny_all_approver() -> Approver

Deny every request.

Source code in src/mcp_for_copilot/approval.py

def deny_all_approver() -> Approver:
    """Deny every request."""

    async def _deny(request: ApprovalRequest) -> ApprovalDecision:
        return ApprovalDecision.deny("deny_all approver", approver="deny_all")

    return _deny

static_approver

static_approver(
    *,
    allow: tuple[str, ...] | list[str] = (),
    deny: tuple[str, ...] | list[str] = (),
) -> Approver

Approve tools in allow, deny tools in deny, deny anything else.

Source code in src/mcp_for_copilot/approval.py

def static_approver(
    *, allow: tuple[str, ...] | list[str] = (), deny: tuple[str, ...] | list[str] = ()
) -> Approver:
    """Approve tools in *allow*, deny tools in *deny*, deny anything else."""
    allowed = set(allow)
    denied = set(deny)

    async def _decide(request: ApprovalRequest) -> ApprovalDecision:
        if request.tool in allowed:
            return ApprovalDecision.allow("static allowlist", approver="static")
        if request.tool in denied:
            return ApprovalDecision.deny("static denylist", approver="static")
        return ApprovalDecision.deny("tool not in the static allowlist", approver="static")

    return _decide

Orchestrator

mcp_for_copilot.orchestrator

Orchestrator — the model -> tool -> model loop.

One run() call performs up to settings.mcp_max_tool_rounds rounds. Each round:

1. Ask the model, passing the tool schemas it is allowed to use.
2. If the model returns no tool calls, the turn is finished.
3. For each requested tool call, run the three gates: a. Allowlist — :class:~mcp_for_copilot.tool_router.ToolRouter b. Approval — :class:~mcp_for_copilot.approval.ApprovalLayer c. Dispatch — :class:~mcp_for_copilot.mcp_client.MCPClient
4. Feed the tool results back to the model and repeat.

A denied or failed tool is reported back to the model as an error result rather than aborting the turn, so the model can recover or explain.

LLMMCPOrchestrator

LLMMCPOrchestrator(
    *,
    provider: LLMProvider | None = None,
    client: MCPClient | MCPClientManager | None = None,
    router: ToolRouter | None = None,
    approval: ApprovalLayer | None = None,
    settings: Settings | None = None,
)

Drives the model/tool loop with the three gates in place.

Source code in src/mcp_for_copilot/orchestrator.py

def __init__(
    self,
    *,
    provider: LLMProvider | None = None,
    client: MCPClient | MCPClientManager | None = None,
    router: ToolRouter | None = None,
    approval: ApprovalLayer | None = None,
    settings: Settings | None = None,
) -> None:
    self.settings = settings or get_settings()
    self.provider = provider or LLMProvider()
    self.client = client
    self.router = router or ToolRouter(
        allowed_tools=self.settings.mcp_allowed_tools or None,
        require_approval=self.settings.mcp_require_approval,
    )
    self.approval = approval or ApprovalLayer()
    # The model's self-reported plan for the current turn. Reset at the
    # start of every ``run()`` so one turn never inherits another's plan.
    self._plan: list[PlanStep] = []
    # Repeated-call detection for the current turn. Also reset per turn.
    self._loop = LoopDetector(
        soft_threshold=self.settings.mcp_loop_soft_threshold,
        hard_threshold=self.settings.mcp_loop_hard_threshold,
    )

run async

run(
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> OrchestratorResult

Run the loop and return the final answer with its audit trail.

Source code in src/mcp_for_copilot/orchestrator.py

async def run(
    self,
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> OrchestratorResult:
    """Run the loop and return the final answer with its audit trail."""
    rounds_cap = max_rounds if max_rounds is not None else self.settings.mcp_max_tool_rounds
    self._plan = []
    self._loop = LoopDetector(
        soft_threshold=self.settings.mcp_loop_soft_threshold,
        hard_threshold=self.settings.mcp_loop_hard_threshold,
    )
    history: list[dict[str, Any]] = [
        {"role": "system", "content": self._system_prompt(system)},
        *messages,
    ]

    result = OrchestratorResult(model=self.provider.model_id)
    tools = await self._available_tool_schemas()

    for round_index in range(rounds_cap + 1):
        result.rounds = round_index + 1
        try:
            response = await self.provider.complete(
                history, tools=tools or None, model_id=model_id
            )
        except LLMProviderError as exc:
            log_warning("orchestrator provider failure: %s", redact(str(exc)))
            result.error = str(exc)
            result.content = f"Provider error: {exc}"
            return result

        result.model = response.model or result.model
        if response.usage:
            result.usage = response.usage

        if not response.wants_tools:
            result.content = response.content
            return result

        if round_index >= rounds_cap:
            log_warning("tool round cap reached (%d)", rounds_cap)
            result.content = (
                response.content
                or f"Stopped after {rounds_cap} tool rounds without a final answer."
            )
            return result

        history.append(self._assistant_tool_message(response))

        for call in response.tool_calls:
            if call.name == PLAN_TOOL_NAME:
                invocation, message = self._handle_plan_call(call)
            else:
                invocation, message = await self._handle_tool_call(call, round_index)
            result.invocations.append(invocation)
            history.append(message)

            # Loop detection: only real tool calls count. The synthetic
            # plan tool is excluded because re-publishing a plan is normal.
            if self.settings.mcp_loop_detection and call.name != PLAN_TOOL_NAME:
                warn, stop, count = self._loop.record(call.name, call.arguments)
                if stop:
                    log_warning(
                        "loop detected: %s repeated %d times; stopping turn",
                        redact(call.name),
                        count,
                    )
                    result.loop_stopped = True
                    result.plan = list(self._plan)
                    result.content = (
                        response.content
                        or f"Stopped: the model repeated the same tool call "
                        f"({call.name}) {count} times without making progress."
                    )
                    return result
                if warn:
                    log_info("loop warning: %s repeated %d times", redact(call.name), count)
                    history.append(
                        {
                            "role": "system",
                            "content": _LOOP_WARNING.format(count=count),
                        }
                    )

        result.plan = list(self._plan)
        reminder = self._plan_message()
        if reminder is not None:
            history.append(reminder)

    return result

LoopDetector

LoopDetector(
    *, soft_threshold: int = 3, hard_threshold: int = 5
)

Count repeated identical tool calls and decide when to warn or stop.

This mirrors the CLI loop detection in Cline (soft threshold 3, hard threshold 5): a soft hit injects a warning so the model can change course, and a hard hit ends the turn instead of letting a stuck model burn every remaining round.

Only identical calls count. A model that calls the same tool with different arguments is making progress, not looping.

Source code in src/mcp_for_copilot/orchestrator.py

def __init__(self, *, soft_threshold: int = 3, hard_threshold: int = 5) -> None:
    self.soft_threshold = max(1, soft_threshold)
    self.hard_threshold = max(self.soft_threshold + 1, hard_threshold)
    self._counts: dict[str, int] = {}
    self._warned: set[str] = set()

repeated property

repeated: dict[str, int]

Signatures seen more than once, with their counts.

record

record(
    name: str, arguments: dict[str, Any]
) -> tuple[bool, bool, int]

Record one call and report (warn, stop, count).

warn is True only on the first soft hit for a signature, so the model is nudged once rather than on every subsequent repeat.

Source code in src/mcp_for_copilot/orchestrator.py

def record(self, name: str, arguments: dict[str, Any]) -> tuple[bool, bool, int]:
    """Record one call and report ``(warn, stop, count)``.

    ``warn`` is ``True`` only on the first soft hit for a signature, so the
    model is nudged once rather than on every subsequent repeat.
    """
    signature = _call_signature(name, arguments)
    count = self._counts.get(signature, 0) + 1
    self._counts[signature] = count
    if count >= self.hard_threshold:
        return False, True, count
    if count >= self.soft_threshold and signature not in self._warned:
        self._warned.add(signature)
        return True, False, count
    return False, False, count

OrchestratorResult dataclass

OrchestratorResult(
    content: str = "",
    model: str = "",
    rounds: int = 0,
    invocations: list[ToolInvocation] = list(),
    usage: dict[str, Any] = dict(),
    error: str = "",
    plan: list[PlanStep] = list(),
    loop_stopped: bool = False,
)

The final answer plus the audit trail of every tool call.

plan_progress property

plan_progress: dict[str, int]

Count of steps per status, for a progress display.

PlanStep dataclass

PlanStep(description: str, status: str = 'pending')

One step of the model's working plan.

The plan is advisory: it is the model's own statement of intent, kept so the caller can show progress and so the model can be reminded of what it said it would do. It is never used to gate a tool call -- the three gates in :meth:LLMMCPOrchestrator._handle_tool_call remain the only authority.

ToolInvocation dataclass

ToolInvocation(
    tool: str,
    arguments: dict[str, Any] = dict(),
    allowed: bool = False,
    approved: bool | None = None,
    executed: bool = False,
    is_error: bool = False,
    result: str = "",
    reason: str = "",
)

A record of one tool call and what happened to it.

Gateway facade

mcp_for_copilot.gateway.facade

Gateway facade — one object that owns the whole stack.

:class:Gateway wires the provider, the MCP client, the router, the approval layer and the orchestrator together, and manages the MCP connection lifetime.

Typical use::

async with Gateway() as gateway:
    result = await gateway.chat([{"role": "user", "content": "List the files"}])
    print(result.content)

Gateway

Gateway(
    *,
    settings: Settings | None = None,
    provider: LLMProvider | None = None,
    approver: Approver | None = None,
    connect_mcp: bool = True,
)

Owns the provider + MCP client + orchestrator for one process.

Source code in src/mcp_for_copilot/gateway/facade.py

def __init__(
    self,
    *,
    settings: Settings | None = None,
    provider: LLMProvider | None = None,
    approver: Approver | None = None,
    connect_mcp: bool = True,
) -> None:
    self.settings = settings or get_settings()
    self.provider = provider or LLMProvider()
    self.debug_marathon = DebugMarathon(self.provider)
    self._approver = approver
    self._connect_mcp = connect_mcp

    # ``client`` is either a single :class:`MCPClient` (deprecated
    # ``MCP_SERVER_URL`` path) or an :class:`MCPClientManager` when
    # ``MCP_SERVERS`` lists more than one server. The orchestrator already
    # branches on the type, so both are valid here.
    self.client: MCPClient | MCPClientManager | None = None
    self.router = ToolRouter(
        allowed_tools=self.settings.mcp_allowed_tools or None,
        require_approval=self.settings.mcp_require_approval,
    )
    self.approval = ApprovalLayer(approver)
    self.orchestrator = LLMMCPOrchestrator(
        provider=self.provider,
        client=None,
        router=self.router,
        approval=self.approval,
        settings=self.settings,
    )
    self._connected = False

chat async

chat(
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> GatewayResult

Run one chat turn through the full tool loop.

Source code in src/mcp_for_copilot/gateway/facade.py

async def chat(
    self,
    messages: list[dict[str, Any]],
    *,
    system: str | None = None,
    model_id: str | None = None,
    max_rounds: int | None = None,
) -> GatewayResult:
    """Run one chat turn through the full tool loop."""
    if not self._connected:
        await self.start()
    result = await self.orchestrator.run(
        messages, system=system, model_id=model_id, max_rounds=max_rounds
    )
    return GatewayResult.from_orchestrator(result)

list_tools async

list_tools() -> list[dict[str, Any]]

Return the allowed tools as OpenAI function schemas.

Source code in src/mcp_for_copilot/gateway/facade.py

async def list_tools(self) -> list[dict[str, Any]]:
    """Return the allowed tools as OpenAI function schemas."""
    if not self._connected:
        await self.start()
    if self.client is None:
        return []
    try:
        if isinstance(self.client, MCPClientManager):
            listed = await self.client.list_all_tools()
            tools = [tool for tools in listed.values() for tool in tools]
        else:
            tools = await self.client.list_tools()
    except MCPClientError as exc:
        log_warning("could not list tools: %s", redact(str(exc)))
        return []
    self.router.register_all(tools)
    return LLMProvider.to_openai_tools(self.router.available())

start async

start() -> None

Connect to the MCP server(s), if any are configured.

MCP_SERVERS (a JSON array) takes precedence and yields an :class:MCPClientManager. When it is unset, the deprecated single MCP_SERVER_URL fields are used exactly as before, so existing deployments keep working unchanged.

Source code in src/mcp_for_copilot/gateway/facade.py

async def start(self) -> None:
    """Connect to the MCP server(s), if any are configured.

    ``MCP_SERVERS`` (a JSON array) takes precedence and yields an
    :class:`MCPClientManager`. When it is unset, the deprecated single
    ``MCP_SERVER_URL`` fields are used exactly as before, so existing
    deployments keep working unchanged.
    """
    if self._connected or not self._connect_mcp:
        return

    if self.settings.mcp_servers_parsed:
        self.client = await self._connect_manager()
    else:
        self.client = await self._connect_single()

    self.orchestrator.client = self.client
    self._connected = True

status

status() -> dict[str, Any]

Non-secret status snapshot, safe to return over HTTP.

Source code in src/mcp_for_copilot/gateway/facade.py

def status(self) -> dict[str, Any]:
    """Non-secret status snapshot, safe to return over HTTP."""
    if isinstance(self.client, MCPClientManager):
        servers = sorted(self.client._clients)
        server_info: dict[str, Any] = {"servers": servers}
        transport = "multi"
    else:
        servers = []
        server_info = self.client.server_info if self.client else {}
        transport = self.settings.mcp_transport
    return {
        "llm": self.provider.describe(),
        "mcp": {
            "transport": transport,
            "configured": self.settings.mcp_configured,
            "connected": self.client is not None,
            "server_info": server_info,
            "servers": servers,
            "allowed_tools": list(self.router.allowed_tools),
            "require_approval": self.settings.mcp_require_approval,
            "approver_configured": self.approval.configured,
        },
    }

GatewayResult dataclass

GatewayResult(
    content: str = "",
    model: str = "",
    rounds: int = 0,
    used_tools: list[str] = list(),
    invocations: list[dict[str, Any]] = list(),
    usage: dict[str, Any] = dict(),
    error: str = "",
    plan: list[dict[str, str]] = list(),
    plan_progress: dict[str, int] = dict(),
    loop_stopped: bool = False,
)

A chat completion plus the tool audit trail.

HTTP application

mcp_for_copilot.gateway.app

FastAPI app — an OpenAI-compatible gateway in front of GPT-6 Sol + MCP.

Endpoints

GET /health Liveness probe. Always 200 when the process is up. GET /v1/models OpenAI-compatible model list. POST /v1/chat/completions OpenAI-compatible chat completions, with optional SSE streaming. GET /v1/status Non-secret configuration snapshot.

Authentication

When GATEWAY_API_KEY is set, every /v1/* request must carry Authorization: Bearer <key>. When it is empty the gateway accepts any key, which is intended for local development only.

create_app

create_app(
    *,
    settings: Settings | None = None,
    gateway: Gateway | None = None,
) -> FastAPI

Build the FastAPI app. gateway is injectable for tests.

Source code in src/mcp_for_copilot/gateway/app.py

def create_app(
    *,
    settings: Settings | None = None,
    gateway: Gateway | None = None,
) -> FastAPI:
    """Build the FastAPI app. *gateway* is injectable for tests."""
    resolved = settings or get_settings()

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        app.state.gateway = gateway or Gateway(settings=resolved)
        await app.state.gateway.start()
        app.state.gateway.log_summary()
        try:
            yield
        finally:
            await app.state.gateway.stop()

    app = FastAPI(
        title="MCP for Copilot",
        version="0.1.0",
        description=(
            "OpenAI-compatible gateway that gives GPT-6 Sol tool access "
            "through the Model Context Protocol."
        ),
        lifespan=lifespan,
    )
    app.add_middleware(
        CORSMiddleware,
        allow_origins=resolved.gateway_cors_origins,
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    router = APIRouter()

    def _check_auth(authorization: str | None) -> None:
        expected = resolved.gateway_api_key.strip()
        if not expected:
            return
        token = ""
        if authorization and authorization.lower().startswith("bearer "):
            token = authorization[7:].strip()
        if token != expected:
            raise HTTPException(status_code=401, detail="Invalid API key")

    def _get_gateway(request: Request) -> Gateway:
        current: Gateway | None = getattr(request.app.state, "gateway", None)
        if current is None:
            raise HTTPException(status_code=503, detail="Gateway is not ready")
        return current

    @router.post("/debug/marathon", status_code=202)
    async def submit_debug_marathon(
        payload: DebugMarathonRequest,
        request: Request,
        authorization: str | None = Header(default=None),
    ) -> dict[str, Any]:
        _check_auth(authorization)
        try:
            return _get_gateway(request).debug_marathon.submit(
                [task.model_dump() for task in payload.tasks]
            )
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc

    @router.get("/debug/marathon/{job_id}")
    async def get_debug_marathon(
        job_id: str,
        request: Request,
        authorization: str | None = Header(default=None),
    ) -> dict[str, Any]:
        _check_auth(authorization)
        job = _get_gateway(request).debug_marathon.get(job_id)
        if job is None:
            raise HTTPException(status_code=404, detail="Debug marathon job not found")
        return job

    @app.get("/health")
    async def health() -> dict[str, Any]:
        return {"status": "ok", "service": "mcp-for-copilot"}

    @router.get("/models")
    async def list_models(
        authorization: str | None = Header(default=None),
    ) -> dict[str, Any]:
        _check_auth(authorization)
        created = int(time.time())
        data = [
            {
                "id": model_id,
                "object": "model",
                "created": created,
                "owned_by": "openai",
                "metadata": {
                    "label": spec["label"],
                    "tier": spec["tier"],
                    "context_window": spec["context_window"],
                    "max_output": spec["max_output"],
                    "reasoning_effort": list(spec["reasoning_effort"]),
                    "aliases": sorted(
                        a for a, target in MODEL_ALIASES.items() if target == model_id
                    ),
                },
            }
            for model_id, spec in sorted(KNOWN_MODELS.items())
        ]
        return {"object": "list", "data": data}

    @router.get("/status")
    async def status(
        request: Request,
        authorization: str | None = Header(default=None),
    ) -> dict[str, Any]:
        _check_auth(authorization)
        return _get_gateway(request).status()

    @router.post("/chat/completions")
    async def chat_completions(
        request: Request,
        authorization: str | None = Header(default=None),
    ) -> Any:
        _check_auth(authorization)
        try:
            body = await request.json()
        except ValueError as exc:
            raise HTTPException(status_code=400, detail="Body must be JSON") from exc
        if not isinstance(body, dict):
            raise HTTPException(status_code=400, detail="Body must be a JSON object")

        messages = body.get("messages")
        if not isinstance(messages, list) or not messages:
            raise HTTPException(status_code=400, detail="'messages' must be a non-empty list")

        try:
            model_id = resolve_model_id(body.get("model"))
        except LLMProviderError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

        gateway_instance = _get_gateway(request)
        system = None
        conversation: list[dict[str, Any]] = []
        for message in messages:
            if not isinstance(message, dict):
                continue
            if message.get("role") == "system":
                system = str(message.get("content") or "")
            else:
                conversation.append(message)

        completion_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
        created = int(time.time())

        if body.get("stream"):
            return StreamingResponse(
                _stream(
                    gateway_instance,
                    conversation,
                    system=system,
                    model_id=model_id,
                    completion_id=completion_id,
                    created=created,
                ),
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )

        try:
            result = await gateway_instance.chat(conversation, system=system, model_id=model_id)
        except LLMProviderError as exc:
            log_warning("chat failed: %s", redact(str(exc)))
            raise HTTPException(status_code=502, detail=str(exc)) from exc

        if result.error:
            log_warning("chat failed: %s", redact(result.error))
            raise HTTPException(status_code=502, detail=result.error)

        return {
            "id": completion_id,
            "object": "chat.completion",
            "created": created,
            "model": result.model or model_id,
            "choices": [
                {
                    "index": 0,
                    "message": {"role": "assistant", "content": result.content},
                    "finish_reason": "stop",
                }
            ],
            "usage": _usage(_messages_to_text(conversation), result.content, result.usage),
            "x_gateway": {
                "rounds": result.rounds,
                "used_tools": result.used_tools,
                "invocations": result.invocations,
                "plan": result.plan,
                "plan_progress": result.plan_progress,
                "loop_stopped": result.loop_stopped,
            },
        }

    app.include_router(router, prefix="/v1")
    return app

Debug marathon

mcp_for_copilot.debug_marathon

Prioritized, sequential debug runs through the GPT-6 model ladder.

DebugMarathon

DebugMarathon(provider: LLMProvider)

Own an in-process FIFO job queue and execute tasks one at a time.

Source code in src/mcp_for_copilot/debug_marathon.py

def __init__(self, provider: LLMProvider) -> None:
    self.provider = provider
    self._jobs: dict[str, dict[str, Any]] = {}
    self._queue: asyncio.Queue[tuple[str, list[dict[str, Any]]]] = asyncio.Queue()
    self._worker: asyncio.Task[None] | None = None

prioritize_tasks

prioritize_tasks(
    tasks: list[dict[str, Any]],
) -> list[dict[str, Any]]

Validate and stably order tasks by priority, difficulty, and complexity.

Source code in src/mcp_for_copilot/debug_marathon.py

def prioritize_tasks(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Validate and stably order tasks by priority, difficulty, and complexity."""
    if not 1 <= len(tasks) <= MAX_TASKS_PER_JOB:
        raise ValueError(f"tasks must contain 1 to {MAX_TASKS_PER_JOB} items")

    normalized: list[dict[str, Any]] = []
    for task in tasks:
        if not isinstance(task, dict):
            raise ValueError("each task must be an object")
        question = str(task.get("question") or "").strip()
        if not question or len(question) > 10000:
            raise ValueError("each task needs a question of 1 to 10000 characters")
        item: dict[str, Any] = {"question": question}
        title = str(task.get("title") or "").strip()
        if title:
            if len(title) > 200:
                raise ValueError("task title must be 200 characters or fewer")
            item["title"] = title
        for field in ("priority", "difficulty", "complexity"):
            score = task.get(field, 3)
            if isinstance(score, bool) or not isinstance(score, int):
                raise ValueError(f"{field} must be an integer from 1 to 5")
            if not 1 <= score <= 5:
                raise ValueError(f"{field} must be an integer from 1 to 5")
            item[field] = score
        normalized.append(item)

    ranked = sorted(
        enumerate(normalized),
        key=lambda pair: (
            -pair[1]["priority"],
            -pair[1]["difficulty"],
            -pair[1]["complexity"],
            pair[0],
        ),
    )
    return [
        {
            **task,
            "task_id": str(uuid4()),
            "queue_position": position,
            "priority_score": (
                task["priority"] * 100 + task["difficulty"] * 10 + task["complexity"]
            ),
        }
        for position, (_, task) in enumerate(ranked, start=1)
    ]

MCP server

mcp_for_copilot.gateway.server

stdio MCP server — exposes the gateway itself as MCP tools.

Run it directly::

python -m mcp_for_copilot.gateway.server

or let a client spawn it::

mcp-for-copilot mcp

Tools

gpt6_chat Send a prompt to GPT-6 Sol and return the answer, plus the working plan the model kept (when plan mode is enabled). gpt6_models List the models this gateway knows about. gpt6_status Report the non-secret gateway configuration. gpt6_tools List the MCP tools the gateway is allowed to call. gpt6_plan Return the working plan from the most recent gpt6_chat call. gpt6_debug_marathon Queue prioritized debug tasks through GPT-6 Luna, Sol, and Astra. gpt6_debug_marathon_status Get a queued debug marathon's progress and results. copilot_install Verify (and optionally install) the GitHub Copilot CLI. hubspot_track Record a lead through VihokAI's own HubSpot route. vihokai_deploy Report on, and optionally push, the VihokAI repository.

build_server

build_server(gateway: Gateway | None = None) -> Server

Create the MCP server. gateway is injectable for tests.

Source code in src/mcp_for_copilot/gateway/server.py

def build_server(gateway: Gateway | None = None) -> Server:
    """Create the MCP server. *gateway* is injectable for tests."""
    state: dict[str, Gateway | None] = {"gateway": gateway}
    # Last plan seen from ``gpt6_chat``, so ``gpt6_plan`` can report it without
    # re-running the model. Read-only: it never influences routing or approval.
    last_plan: dict[str, Any] = {"plan": [], "plan_progress": {}}

    async def _gateway() -> Gateway:
        current = state["gateway"]
        if current is None:
            current = Gateway(settings=get_settings())
            await current.start()
            state["gateway"] = current
        return current

    async def list_tools(_ctx: Any, _params: Any = None) -> ListToolsResult:
        return ListToolsResult(
            tools=[
                Tool(
                    name="gpt6_chat",
                    description=(
                        "Ask GPT-6 Sol a question. The model may call MCP tools "
                        "through the gateway to answer."
                    ),
                    input_schema=_CHAT_SCHEMA,
                ),
                Tool(
                    name="gpt6_models",
                    description="List the models this gateway knows about, with aliases.",
                    input_schema=_EMPTY_SCHEMA,
                ),
                Tool(
                    name="gpt6_status",
                    description="Report the gateway configuration (no secrets).",
                    input_schema=_EMPTY_SCHEMA,
                ),
                Tool(
                    name="gpt6_tools",
                    description="List the MCP tools the gateway is allowed to call.",
                    input_schema=_EMPTY_SCHEMA,
                ),
                Tool(
                    name="gpt6_plan",
                    description=(
                        "Return the working plan from the most recent gpt6_chat call. "
                        "Empty unless plan mode is enabled. Advisory only."
                    ),
                    input_schema=_EMPTY_SCHEMA,
                ),
                Tool(
                    name="gpt6_debug_marathon",
                    description=(
                        "Queue 1-10 debug tasks. Tasks are ordered by priority, "
                        "difficulty, then complexity (highest first). Each task "
                        "runs through GPT-6 Luna, then Sol, then Astra. Returns a job_id."
                    ),
                    input_schema=_DEBUG_MARATHON_SCHEMA,
                ),
                Tool(
                    name="gpt6_debug_marathon_status",
                    description="Get progress and results for a debug marathon job_id.",
                    input_schema=_DEBUG_MARATHON_STATUS_SCHEMA,
                ),
                Tool(
                    name="copilot_install",
                    description=(
                        "Verify the GitHub Copilot CLI, and optionally install it with "
                        "`npm install -g @github/copilot`. Verifies only unless install=true."
                    ),
                    input_schema=_COPILOT_INSTALL_SCHEMA,
                ),
                Tool(
                    name="hubspot_track",
                    description=(
                        "Record a lead through VihokAI's /api/hubspot/conversation route, "
                        "which holds the HubSpot credential server-side. Reports whether the "
                        "app has HubSpot configured."
                    ),
                    input_schema=_HUBSPOT_TRACK_SCHEMA,
                ),
                Tool(
                    name="vihokai_deploy",
                    description=(
                        "Report the VihokAI repo branch, HEAD, commits ahead of origin, and "
                        "dirty state. Pushes only when push=true; refuses on main without "
                        "allow_main=true."
                    ),
                    input_schema=_VIHOKAI_DEPLOY_SCHEMA,
                ),
            ]
        )

    async def call_tool(_ctx: Any, params: Any) -> CallToolResult:
        name = str(getattr(params, "name", "") or "")
        args = dict(getattr(params, "arguments", None) or {})
        blocks: list[Any] = list(await _dispatch(name, args))
        return CallToolResult(content=blocks)

    async def _dispatch(name: str, args: dict[str, Any]) -> list[TextContent]:
        try:
            if name == "gpt6_chat":
                prompt = str(args.get("prompt") or "").strip()
                if not prompt:
                    return [TextContent(type="text", text="Error: 'prompt' is required")]
                current = await _gateway()
                result = await current.chat(
                    [{"role": "user", "content": prompt}],
                    system=args.get("system"),
                    model_id=args.get("model"),
                )
                payload = {
                    "content": result.content,
                    "model": result.model,
                    "rounds": result.rounds,
                    "used_tools": result.used_tools,
                    "plan": result.plan,
                    "plan_progress": result.plan_progress,
                    "loop_stopped": result.loop_stopped,
                }
                last_plan["plan"] = result.plan
                last_plan["plan_progress"] = result.plan_progress
                return [TextContent(type="text", text=json.dumps(payload, ensure_ascii=False))]

            if name == "gpt6_models":
                payload = {
                    model_id: {
                        "label": spec["label"],
                        "tier": spec["tier"],
                        "context_window": spec["context_window"],
                        "max_output": spec["max_output"],
                        "aliases": sorted(
                            a for a, target in MODEL_ALIASES.items() if target == model_id
                        ),
                    }
                    for model_id, spec in sorted(KNOWN_MODELS.items())
                }
                return [TextContent(type="text", text=json.dumps(payload, ensure_ascii=False))]

            if name == "gpt6_status":
                current = await _gateway()
                return [
                    TextContent(
                        type="text",
                        text=json.dumps(current.status(), ensure_ascii=False),
                    )
                ]

            if name == "gpt6_tools":
                current = await _gateway()
                tools = await current.list_tools()
                return [
                    TextContent(
                        type="text",
                        text=json.dumps([t["function"]["name"] for t in tools], ensure_ascii=False),
                    )
                ]

            if name == "gpt6_plan":
                return [
                    TextContent(
                        type="text",
                        text=json.dumps(last_plan, ensure_ascii=False),
                    )
                ]

            if name == "gpt6_debug_marathon":
                current = await _gateway()
                submitted_job = current.debug_marathon.submit(args.get("tasks") or [])
                return [
                    TextContent(type="text", text=json.dumps(submitted_job, ensure_ascii=False))
                ]

            if name == "gpt6_debug_marathon_status":
                current = await _gateway()
                status_payload = current.debug_marathon.get(str(args.get("job_id") or ""))
                if status_payload is None:
                    return [TextContent(type="text", text="Error: debug marathon job not found")]
                return [
                    TextContent(type="text", text=json.dumps(status_payload, ensure_ascii=False))
                ]

            if name == "copilot_install":
                return [TextContent(type="text", text=json.dumps(await _copilot_install(args), ensure_ascii=False))]

            if name == "hubspot_track":
                return [TextContent(type="text", text=json.dumps(await _hubspot_track(args), ensure_ascii=False))]

            if name == "vihokai_deploy":
                return [TextContent(type="text", text=json.dumps(await _vihokai_deploy(args), ensure_ascii=False))]

            return [TextContent(type="text", text=f"Error: unknown tool {name!r}")]
        except Exception as exc:
            log_error("mcp tool %s failed: %s", redact(name), redact(str(exc)))
            return [TextContent(type="text", text=f"Error: {exc}")]

    return Server(
        SERVER_NAME,
        on_list_tools=list_tools,
        on_call_tool=call_tool,
    )

main

main() -> None

Console entry point for python -m mcp_for_copilot.gateway.server.

Source code in src/mcp_for_copilot/gateway/server.py

def main() -> None:
    """Console entry point for ``python -m mcp_for_copilot.gateway.server``."""
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(_run())

Logging utilities

mcp_for_copilot.logging_utils

Secret-safe logging.

Every log line produced by this package goes through :func:safe_log, which runs the message through :func:redact first. The redactor is deliberately aggressive: it removes anything that looks like a credential even when the caller forgot to mark it as one.

Threat model

A tool result or an upstream error body can contain a token that the model echoed back, or a connection string with an embedded password. If that text reaches a log file it is a leak, so redaction happens at the boundary rather than at each call site.

contains_secret

contains_secret(text: str) -> bool

True when text still contains something the redactor would remove.

Used by the test-suite to assert that no secret reaches a log sink.

Source code in src/mcp_for_copilot/logging_utils.py

def contains_secret(text: str) -> bool:
    """True when *text* still contains something the redactor would remove.

    Used by the test-suite to assert that no secret reaches a log sink.
    """
    if not text:
        return False
    for pattern, _ in _REDACTION_RULES:
        for match in pattern.finditer(text):
            candidate = match.group(0)
            # Ignore matches that are already redaction markers or placeholders.
            if "REDACTED" in candidate:
                continue
            if _PLACEHOLDER_RE.match(candidate):
                continue
            return True
    return False

redact

redact(value: Any) -> str

Return value as a string with every credential-looking substring removed.

Safe to call on any object; None becomes "".

Source code in src/mcp_for_copilot/logging_utils.py

def redact(value: Any) -> str:
    """Return *value* as a string with every credential-looking substring removed.

    Safe to call on any object; ``None`` becomes ``""``.
    """
    if value is None:
        return ""
    text = value if isinstance(value, str) else str(value)
    for pattern, replacement in _REDACTION_RULES:
        text = pattern.sub(replacement, text)
    return text

redact_mapping

redact_mapping(data: dict[str, Any]) -> dict[str, Any]

Redact every value in a shallow mapping (keys are preserved).

Source code in src/mcp_for_copilot/logging_utils.py

def redact_mapping(data: dict[str, Any]) -> dict[str, Any]:
    """Redact every value in a shallow mapping (keys are preserved)."""
    return {k: redact(v) for k, v in data.items()}

safe_log

safe_log(
    level: int, message: str, *args: Any, **kwargs: Any
) -> None

Log message after redaction.

%-style args are redacted individually so a secret passed as an argument cannot slip through the format string. Scalar args keep their type so %d/%f specifiers still work.

Source code in src/mcp_for_copilot/logging_utils.py

def safe_log(level: int, message: str, *args: Any, **kwargs: Any) -> None:
    """Log *message* after redaction.

    ``%``-style args are redacted individually so a secret passed as an
    argument cannot slip through the format string. Scalar args keep their
    type so ``%d``/``%f`` specifiers still work.
    """
    safe_args = tuple(_redact_arg(a) for a in args)
    logger.log(level, redact(message), *safe_args, **kwargs)

CLI

mcp_for_copilot.cli

Command-line interface for the gateway.

Subcommands

serve Run the FastAPI gateway with uvicorn. mcp Run the stdio MCP server. chat Send one prompt and print the answer. marathon Queue debug tasks and poll the job until it finishes. models List the known models. status Print the non-secret configuration.

main

main(argv: list[str] | None = None) -> int

Entry point for the mcp-for-copilot console script.

Source code in src/mcp_for_copilot/cli.py

def main(argv: list[str] | None = None) -> int:
    """Entry point for the ``mcp-for-copilot`` console script."""
    args = _build_parser().parse_args(argv)
    handlers = {
        "serve": _cmd_serve,
        "mcp": _cmd_mcp,
        "chat": _cmd_chat,
        "marathon": _cmd_marathon,
        "models": _cmd_models,
        "status": _cmd_status,
    }
    return handlers[args.command](args)