Coverage for src/quickmcp/cli.py: 39%

1#!/usr/bin/env python 

2""" 

3QuickMCP CLI - Command line interface for QuickMCP 

4""" 

5 

6import argparse 

7import sys 

8import json 

9from pathlib import Path 

10from typing import Optional, List 

11 

12from .registry import ServerRegistry, ServerRegistration, register_server 

13from .autodiscovery import discover_servers 

14 

15 

16def cmd_register(args): 

17 """Register a QuickMCP server.""" 

18 command = args.command.split() if isinstance(args.command, str) else args.command 

19 

20 register_server( 

21 name=args.name, 

22 command=command, 

23 description=args.description or "", 

24 working_dir=args.working_dir, 

25 tool_prefix=args.tool_prefix 

26 ) 

27 print(f"✓ Registered server: {args.name}") 

28 

29 

30def cmd_unregister(args): 

31 """Unregister a QuickMCP server.""" 

32 registry = ServerRegistry() 

33 if registry.unregister(args.name): 

34 print(f"✓ Unregistered server: {args.name}") 

35 else: 

36 print(f"✗ Server not found: {args.name}") 

37 sys.exit(1) 

38 

39 

40def cmd_list(args): 

41 """List registered QuickMCP servers.""" 

42 registry = ServerRegistry() 

43 servers = registry.list() 

44 

45 if not servers: 

46 print("No servers registered.") 

47 return 

48 

49 print(f"Registered QuickMCP servers ({len(servers)}):") 

50 print("-" * 60) 

51 

52 for server in servers: 

53 print(f"\n{server.name}") 

54 print(f" Description: {server.description}") 

55 print(f" Command: {' '.join(server.command)}") 

56 if server.working_dir: 

57 print(f" Working Dir: {server.working_dir}") 

58 if server.tool_prefix: 

59 print(f" Tool Prefix: {server.tool_prefix}") 

60 if server.capabilities: 

61 tools = server.capabilities.get("tools", []) 

62 resources = server.capabilities.get("resources", []) 

63 prompts = server.capabilities.get("prompts", []) 

64 if tools: 

65 print(f" Tools: {', '.join(tools[:3])}{' ...' if len(tools) > 3 else ''}") 

66 if resources: 

67 print(f" Resources: {', '.join(resources[:3])}{' ...' if len(resources) > 3 else ''}") 

68 if prompts: 

69 print(f" Prompts: {', '.join(prompts[:3])}{' ...' if len(prompts) > 3 else ''}") 

70 

71 

72def cmd_discover(args): 

73 """Discover QuickMCP servers.""" 

74 print("Discovering QuickMCP servers...") 

75 

76 # Discover in filesystem 

77 if args.scan_filesystem: 

78 registry = ServerRegistry() 

79 search_paths = [Path(p) for p in args.paths] if args.paths else None 

80 discovered = registry.auto_discover(search_paths) 

81 

82 print(f"\nFound {len(discovered)} servers in filesystem:") 

83 for server in discovered: 

84 print(f" - {server.name}: {server.description}") 

85 if args.auto_register: 

86 registry.register(server) 

87 print(f" ✓ Registered") 

88 

89 # Discover on network 

90 if args.scan_network: 

91 import asyncio 

92 

93 async def discover(): 

94 return await discover_servers(timeout=args.timeout) 

95 

96 network_servers = asyncio.run(discover()) 

97 

98 print(f"\nFound {len(network_servers)} servers on network:") 

99 for server in network_servers: 

100 print(f" - {server.name} ({server.transport})") 

101 print(f" Host: {server.host}:{server.port}") 

102 print(f" Description: {server.description}") 

103 

104 

105def cmd_export(args): 

106 """Export registry as Gleitzeit configuration.""" 

107 registry = ServerRegistry() 

108 config = registry.to_gleitzeit_config() 

109 

110 if args.format == "yaml": 

111 import yaml 

112 output = yaml.dump(config, default_flow_style=False) 

113 else: 

114 output = json.dumps(config, indent=2) 

115 

116 if args.output: 

117 with open(args.output, 'w') as f: 

118 f.write(output) 

119 print(f"✓ Exported configuration to {args.output}") 

120 else: 

121 print(output) 

122 

123 

124def cmd_info(args): 

125 """Show information about a registered server.""" 

126 registry = ServerRegistry() 

127 server = registry.get(args.name) 

128 

129 if not server: 

130 print(f"✗ Server not found: {args.name}") 

131 sys.exit(1) 

132 

133 print(f"Server: {server.name}") 

134 print(f"Description: {server.description}") 

135 print(f"Command: {' '.join(server.command)}") 

136 

137 if server.working_dir: 

138 print(f"Working Directory: {server.working_dir}") 

139 

140 if server.tool_prefix: 

141 print(f"Tool Prefix: {server.tool_prefix}") 

142 

143 if server.capabilities: 

144 print("\nCapabilities:") 

145 for key, value in server.capabilities.items(): 

146 if isinstance(value, list): 

147 print(f" {key}: {len(value)} items") 

148 for item in value[:5]: 

149 print(f" - {item}") 

150 if len(value) > 5: 

151 print(f" ... and {len(value) - 5} more") 

152 else: 

153 print(f" {key}: {value}") 

154 

155 if server.metadata: 

156 print("\nMetadata:") 

157 for key, value in server.metadata.items(): 

158 print(f" {key}: {value}") 

159 

160 

161def main(): 

162 """Main CLI entry point.""" 

163 parser = argparse.ArgumentParser( 

164 description="QuickMCP CLI - Manage and discover MCP servers" 

165 ) 

166 

167 subparsers = parser.add_subparsers(dest="command", help="Commands") 

168 

169 # Register command 

170 register_parser = subparsers.add_parser("register", help="Register a server") 

171 register_parser.add_argument("name", help="Server name") 

172 register_parser.add_argument("command", help="Command to run the server") 

173 register_parser.add_argument("-d", "--description", help="Server description") 

174 register_parser.add_argument("-w", "--working-dir", help="Working directory") 

175 register_parser.add_argument("-p", "--tool-prefix", help="Tool prefix for Gleitzeit") 

176 register_parser.set_defaults(func=cmd_register) 

177 

178 # Unregister command 

179 unregister_parser = subparsers.add_parser("unregister", help="Unregister a server") 

180 unregister_parser.add_argument("name", help="Server name") 

181 unregister_parser.set_defaults(func=cmd_unregister) 

182 

183 # List command 

184 list_parser = subparsers.add_parser("list", help="List registered servers") 

185 list_parser.set_defaults(func=cmd_list) 

186 

187 # Discover command 

188 discover_parser = subparsers.add_parser("discover", help="Discover servers") 

189 discover_parser.add_argument( 

190 "--scan-filesystem", action="store_true", 

191 help="Scan filesystem for QuickMCP servers" 

192 ) 

193 discover_parser.add_argument( 

194 "--scan-network", action="store_true", 

195 help="Scan network for running servers" 

196 ) 

197 discover_parser.add_argument( 

198 "--auto-register", action="store_true", 

199 help="Automatically register discovered servers" 

200 ) 

201 discover_parser.add_argument( 

202 "--paths", nargs="+", 

203 help="Paths to search for servers" 

204 ) 

205 discover_parser.add_argument( 

206 "--timeout", type=float, default=5.0, 

207 help="Network discovery timeout in seconds" 

208 ) 

209 discover_parser.set_defaults(func=cmd_discover) 

210 

211 # Export command 

212 export_parser = subparsers.add_parser("export", help="Export configuration") 

213 export_parser.add_argument( 

214 "-o", "--output", help="Output file (defaults to stdout)" 

215 ) 

216 export_parser.add_argument( 

217 "-f", "--format", choices=["json", "yaml"], default="yaml", 

218 help="Output format" 

219 ) 

220 export_parser.set_defaults(func=cmd_export) 

221 

222 # Info command 

223 info_parser = subparsers.add_parser("info", help="Show server information") 

224 info_parser.add_argument("name", help="Server name") 

225 info_parser.set_defaults(func=cmd_info) 

226 

227 args = parser.parse_args() 

228 

229 if not args.command: 

230 parser.print_help() 

231 sys.exit(1) 

232 

233 # Execute the command 

234 try: 

235 args.func(args) 

236 except KeyboardInterrupt: 

237 print("\nInterrupted.") 

238 sys.exit(1) 

239 except Exception as e: 

240 print(f"Error: {e}", file=sys.stderr) 

241 sys.exit(1) 

242 

243 

244if __name__ == "__main__": 

245 main()