initial MCP support

Author: karmab

extras/chatbot.py

@@ -0,0 +1,61 @@
+import asyncio
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from langchain_openai import ChatOpenAI
+from langchain_mcp_adapters.tools import load_mcp_tools
+from langgraph.prebuilt import create_react_agent
+
+import sys
+
+model = ChatOpenAI(model='granite3.2:latest', base_url='http://127.0.0.1:4000/v1', api_key='xxx')
+
+server_params = StdioServerParameters(command="kmcp.py")
+
+
+async def main_chat():
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            tools = await load_mcp_tools(session)
+            agent = create_react_agent(model, tools)
+
+            print("Terminal chatbot is ready. Type 'exit' to quit.")
+            chat_history = []
+
+            while True:
+                user_input = input("You: ").strip()
+                if user_input.lower() in {"exit", "quit"}:
+                    print("Exiting chatbot.")
+                    break
+
+                chat_history.append({"role": "user", "content": user_input})
+
+                while True:
+                    result = await agent.ainvoke({"messages": chat_history})
+                    last_message = result["messages"][-1]
+
+                    # Check if it's a tool_use message
+                    if hasattr(last_message, 'type') and last_message.type == "tool_use":
+                        # Agent wants to call a tool → MCP server handles it automatically
+                        tool_name = last_message.tool_call[0]['name']
+                        tool_args = last_message.tool_call[0]['args']
+                        print(f"[Agent is calling tool '{tool_name}' with args {tool_args}]")
+
+                        # Append tool_use message to chat history so agent remembers it
+                        chat_history.append({"role": "user", "content": f"Calling tool: {tool_name}"})
+                        continue  # Run agent again to get tool result processed
+
+                    # Otherwise — final bot response
+                    bot_reply = last_message.content
+                    print(f"Bot: {bot_reply}")
+                    chat_history.append({"role": "assistant", "content": bot_reply})
+                    break
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main_chat())
+    except KeyboardInterrupt:
+        print("\nChatbot interrupted. Exiting.")
+        sys.exit(0)

kvirt/mcp.py

@@ -0,0 +1,74 @@
+from mcp.server.fastmcp import FastMCP
+from kvirt.config import Kbaseconfig, Kconfig
+from urllib.request import urlopen
+
+mcp = FastMCP("kclimcp")
+
+
+@mcp.prompt()
+def prompt() -> str:
+    """Indicates contexts of questions related to kcli"""
+    return """You are a helpful assistant who knows everything about kcli, a powerful client and library written
+    in Python and meant to interact with different virtualization providers, easily deploy and customize VMs or
+    full kubernetes/OpenShift clusters. All information about kcli is available at
+    https://github.com/karmab/kcli/blob/main/docs/index.md"""
+
+
+@mcp.resource("resource://kcli-doc.md")
+def get_doc() -> str:
+    """Provides kcli documentation"""
+    url = 'https://raw.githubusercontent.com/karmab/kcli/refs/heads/main/docs/index.md'
+    return urlopen(url).read().decode('utf-8')
+
+
+@mcp.tool()
+def about_kcli() -> str:
+    """What is kcli"""
+    return open('about.txt').read()
+
+
+@mcp.tool()
+def list_clients() -> list:
+    """List kcli clients/providers"""
+    clientstable = ["Client", "Type", "Enabled", "Current"]
+    baseconfig = Kbaseconfig()
+    for client in sorted(baseconfig.clients):
+        enabled = baseconfig.ini[client].get('enabled', True)
+        _type = baseconfig.ini[client].get('type', 'kvm')
+        if client == baseconfig.client:
+            clientstable.append([client, _type, enabled, 'X'])
+        else:
+            clientstable.append([client, _type, enabled, ''])
+    return clientstable
+
+
+@mcp.tool()
+def list_vms(client: str = None) -> list:
+    """List kcli vms for specific client or for default one when unspecified"""
+    return Kconfig(client).k.list()
+
+
+@mcp.tool()
+def info_vm(name: str, client: str = None) -> dict:
+    """Get info of a kcli vm"""
+    return Kconfig(client).k.info(name)
+
+
+@mcp.tool()
+def create_vm(name: str, profile: str, overrides: dict, client: str = None) -> dict:
+    """Create a kcli vm"""
+    return Kconfig(client).create_vm(name, profile, overrides=overrides)
+
+
+@mcp.tool()
+def delete_vm(vm: str, client: str = None) -> dict:
+    """Delete a kcli vm"""
+    return Kconfig(client).k.delete(vm)
+
+
+def main():
+    mcp.run(transport="stdio")
+
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -57,6 +57,7 @@
         kcli=kvirt.cli:cli
         kweb=kvirt.web.main:run
         klist.py=kvirt.klist:main
+        kmcp.py=kvirt.mcp:main
         ksushy=kvirt.ksushy.main:run
         ignitionmerger=kvirt.ignitionmerger:cli
         ekstoken=kvirt.ekstoken:cli