main.py

import argparse
import logging
from typing import Callable, Literal, Annotated
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from mcp.server.fastmcp import FastMCP
from copr.v3 import Client


class Project(BaseModel):
    id: int
    web_url: str
    ownername: str
    name: str
    full_name: str


class BuildStatus(BaseModel):
    id: int
    state: str
    name: str | None = None


class Build(BaseModel):
    id: int
    web_url: str
    state: str
    submitter: str


class BuildFromDistGit(BaseModel):
    """
    Use this schema when you want to build a package Fedora or any other
    DistGit instance
    """
    source_type: Literal["distgit"] = "distgit"
    packagename: str
    namespace: str | None = None


class BuildFromPyPI(BaseModel):
    """
    Use this schema when you want to build a package from PyPI
    """
    source_type: Literal["pypi"] = "pypi"
    packagename: str
    spec_template: str | None = None


# We need to annotate this with a discriminator otherwise AI sometimes doesn't
# know what type to use and therefore uses the first one.
# class to use
BuildSource = Annotated[
    BuildFromDistGit | BuildFromPyPI,
    Field(discriminator="source_type"),
]


logging.basicConfig(level=logging.WARNING)
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)


def copr_create_project(
    ownername: str,
    projectname: str,
    chroots: list[str],
) -> Project:
    """
    Create a Copr project with a given name for a specified owner.
    When creating a new project, at least one chroot must be specified. For
    example `fedora-rawhide-x86_64`
    """
    log.debug("copr_create_project: %s/%s", ownername, projectname)
    client = Client.create_from_config_file()
    project = client.project_proxy.add(ownername, projectname, chroots)

    # Taken from copr-cli action_create
    # This should be either part of python-copr or returned by the API
    owner_part = project.ownername.replace("@", "g/")
    web_url = "/".join([
        client.config["copr_url"].strip("/"),
        "coprs", owner_part, project.name, "",
    ])

    return Project(
        id=project.id,
        web_url=web_url,
        ownername=project.ownername,
        name=project.name,
        full_name=project.full_name,
    )


def copr_build_status(build_id: int) -> BuildStatus:
    """
    Get the status of a Copr build by its ID.
    """
    log.debug("copr_build_status: %s", build_id)
    client = Client.create_from_config_file()
    build = client.build_proxy.get(build_id)
    return BuildStatus(
        id=build.id,
        state=build.state,
    )


def copr_list_builds(ownername: str, projectname: str) -> list[BuildStatus]:
    """
    Get the status of all builds in a Copr project identified by its
    ownername/projectname.
    """
    log.debug("copr_list_builds: %s/%s", ownername, projectname)
    client = Client.create_from_config_file()
    builds = client.build_proxy.get_list(ownername, projectname)
    return [
        BuildStatus(
            id=build.id,
            state=build.state,
            name=build.source_package["name"],
        )
        for build in builds
    ]


def copr_submit_build(
    ownername: str,
    projectname: str,
    source: BuildSource,
) -> Build:
    """
    Submit a new build into a Copr project defined by its ownername and
    projectname. Copr supports multiple source types, see the documentation
    https://docs.copr.fedorainfracloud.org/user_documentation.html#build-source-types
    """
    log.debug(
        "copr_submit_build: %s/%s %s",
        ownername, projectname, source.__class__.__name__,
    )
    client = Client.create_from_config_file()
    match source:
        case BuildFromDistGit():
            build = client.build_proxy.create_from_distgit(
                ownername,
                projectname,
                source.packagename,
                namespace=source.namespace,

            )
        case BuildFromPyPI():
            build = client.build_proxy.create_from_pypi(
                ownername,
                projectname,
                source.packagename,
                spec_template=source.spec_template,
            )

    web_url = "/".join([
        client.config["copr_url"].strip("/"),
        "coprs/build",
        str(build.id),
    ])

    return Build(
        id=build.id,
        web_url=web_url,
        state=build.state,
        submitter=build.submitter,
    )


def copr_enable_repository(ownername: str, projectname: str) -> str:
    """
    Provide instructions for enabling a Copr repository on the user system.
    This requires root privileges and must be run manually by the user.
    """
    return (
        f"This action requires root privileges and therefore requires a manual"
        f"step from the user.\n"
        f"To enable this Copr repository run the following command:\n\n"
        f"    sudo dnf copr enable {ownername}/{projectname}"
    )


def run_mcp(tools: list[Callable], args):
    mcp = FastMCP("copr")
    for tool in tools:
        mcp.add_tool(tool)
    mcp.run()


def run_prompt(tools: list[Callable], args):
    instructions = (
        "You help manage Copr builds. Use tools to get real information.",
    )
    agent = Agent(
        "anthropic:claude-opus-4-6",
        instructions=instructions,
    )
    for tool in tools:
        agent.tool_plain(tool)
    result = agent.run_sync(args.prompt)
    print(result.output)


def main():
    parser = argparse.ArgumentParser(description="Copr AI assistant")
    parser.add_argument(
        "--prompt",
        help="Don't run MCP and send a prompt directly",
    )
    args = parser.parse_args()
    client = Client.create_from_config_file()
    tools = [
        copr_build_status,
        copr_list_builds,
        copr_create_project,
        copr_submit_build,
        copr_enable_repository,
        # We probably don't have to implement wrappers around every python-copr
        # function. We can simply register the client methods like this.
        client.base_proxy.auth_check,
    ]
    if args.prompt:
        run_prompt(tools, args)
    else:
        run_mcp(tools, args)


if __name__ == "__main__":
    main()