algoforge-mcp-server/server.py

from fastmcp import FastMCP

from forgeclient import load_config
from handlers import repos, contents, branches, collab, search

mcp = FastMCP(
    "MQL5 Algo Forge Server",
    instructions=(
        "This server manages MQL5 Algo Forge repositories (a Git platform for "
        "algorithmic trading code) on the user's behalf. "
        "Call get_me first to confirm the connection and learn the user's login — "
        "it is the {owner} for their own repositories. "
        "Use list_repos to discover existing projects before creating a new one. "
        "When committing code, write a clear commit message describing the change. "
        "For multi-step changes, create a branch with create_branch, commit to it, "
        "then open a pull request with open_pull_request rather than committing "
        "straight to the default branch. "
        "Confirm with the user before creating repositories, committing files, "
        "opening pull requests, or tagging releases."
    ),
)


def _ensure():
    """Check that an Algo Forge token is configured; return error dict if not."""
    config = load_config()
    if not config.get("token"):
        import os
        if not os.environ.get("FORGE_TOKEN"):
            return {"error": "No Algo Forge token configured. Set token in "
                             "config.json or the FORGE_TOKEN environment variable."}
    return None


# ── Account / Identity ───────────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def get_me() -> dict:
    """Get the authenticated Algo Forge user — login, full name, and profile URL.
    Call this first to verify the connection and to learn the username, which is
    the {owner} for the user's own repositories."""
    err = _ensure()
    if err:
        return err
    return repos.get_me()


# ── Repositories ─────────────────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def list_repos(limit: int = 30) -> list:
    """List the authenticated user's repositories. Returns name, description,
    default branch, visibility, and clone URL for each. Use this to discover
    existing projects before creating or committing to one. Limit range: 1-50."""
    err = _ensure()
    if err:
        return err
    return repos.list_repos(limit)


@mcp.tool()
def create_repo(
    name: str,
    description: str = "",
    private: bool = False,
    auto_init: bool = True,
    default_branch: str = "main",
    gitignores: str = "",
    readme: str = "",
) -> dict:
    """Create a new repository for the authenticated user. With auto_init=True the
    repo starts with an initial commit (and an optional README/.gitignore template),
    so you can commit files to it immediately. Set private=True for a private repo.
    gitignores is a template name (e.g. "VisualStudio"); readme is "Default"."""
    err = _ensure()
    if err:
        return err
    return repos.create_repo(name, description, private, auto_init,
                             default_branch, gitignores, readme)


# ── File Contents ────────────────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def get_file(owner: str, repo: str, path: str, ref: str | None = None) -> dict:
    """Read a file's text content from a repository. path is relative to the repo
    root (e.g. "Experts/MyEA.mq5"). ref is an optional branch, tag, or commit;
    defaults to the repository's default branch. Returns the decoded text and the
    file's sha (needed to update it later)."""
    err = _ensure()
    if err:
        return err
    return contents.get_file(owner, repo, path, ref)


@mcp.tool()
def commit_file(
    owner: str,
    repo: str,
    path: str,
    content: str,
    message: str,
    branch: str | None = None,
    sha: str | None = None,
) -> dict:
    """Create or update a file in a repository with a single commit. Provide the
    full file text in content and a descriptive commit message. The handler detects
    whether the file already exists and updates it in place if so. path is relative
    to the repo root. branch defaults to the repository's default branch. Returns
    the resulting commit sha."""
    err = _ensure()
    if err:
        return err
    return contents.commit_file(owner, repo, path, content, message, branch, sha)


# ── Branches ─────────────────────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def list_branches(owner: str, repo: str) -> list:
    """List all branches in a repository with their latest commit sha. Use before
    creating a branch or opening a pull request to see what already exists."""
    err = _ensure()
    if err:
        return err
    return branches.list_branches(owner, repo)


@mcp.tool()
def create_branch(
    owner: str,
    repo: str,
    new_branch: str,
    from_branch: str | None = None,
) -> dict:
    """Create a new branch in a repository. from_branch is the source branch to
    branch off (defaults to the repository's default branch). Use this to start a
    feature or fix before committing changes, so they can be reviewed via a pull
    request instead of going straight to the main branch."""
    err = _ensure()
    if err:
        return err
    return branches.create_branch(owner, repo, new_branch, from_branch)


# ── Issues & Pull Requests ───────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def list_issues(owner: str, repo: str, state: str = "open") -> list:
    """List issues in a repository. state is "open", "closed", or "all".
    Pull requests are excluded — only true issues are returned."""
    err = _ensure()
    if err:
        return err
    return collab.list_issues(owner, repo, state)


@mcp.tool()
def create_issue(owner: str, repo: str, title: str, body: str = "") -> dict:
    """Open a new issue in a repository to track a bug, task, or idea. Returns the
    issue number and URL. Useful for logging problems found while reviewing code."""
    err = _ensure()
    if err:
        return err
    return collab.create_issue(owner, repo, title, body)


@mcp.tool()
def open_pull_request(
    owner: str,
    repo: str,
    title: str,
    head: str,
    base: str,
    body: str = "",
) -> dict:
    """Open a pull request to merge the head branch into the base branch. Use after
    committing changes to a feature branch (head) so they can be reviewed before
    merging into the main branch (base). Returns the PR number and whether it is
    mergeable."""
    err = _ensure()
    if err:
        return err
    return collab.open_pull_request(owner, repo, title, head, base, body)


# ── Releases ─────────────────────────────────────────────────────


@mcp.tool()
def create_release(
    owner: str,
    repo: str,
    tag_name: str,
    name: str | None = None,
    body: str = "",
    target: str = "main",
    prerelease: bool = False,
) -> dict:
    """Tag a version of the project. An Algo Forge release is a Git tag with a name
    and release notes (MQL5 has no package format, so this marks versions rather
    than publishing a package). target is the branch or commit to tag. Returns the
    release id and URL."""
    err = _ensure()
    if err:
        return err
    return collab.create_release(owner, repo, tag_name, name, body,
                                 target, False, prerelease)


# ── Search / Explore ─────────────────────────────────────────────


@mcp.tool(annotations={"readOnlyHint": True})
def search_repos(query: str, limit: int = 10) -> list:
    """Search public projects across Algo Forge (the Explore section). Use to find
    open-source MQL5 code, study other developers' projects, or check whether a
    project name is taken. Limit range: 1-50."""
    err = _ensure()
    if err:
        return err
    return search.search_repos(query, limit)


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