1+ import asyncio
2+ import os
3+ import shutil
4+ import sys
5+ import json
6+ import yaml
7+
8+ from fastmcp import Client
9+ from fastmcp .client .transports import StdioTransport
10+
11+ from prompt_toolkit import PromptSession
12+
13+ async def _print_tool_list (tool_map : dict ):
14+ """Prints the list of available tools in a pretty format."""
15+ print ("\n Available tools:" )
16+ for tool_name in sorted (tool_map .keys ()):
17+ tool_info = tool_map [tool_name ]
18+ print (f" - { tool_info .name } : { tool_info .description } " )
19+
20+ async def main ():
21+ # Get the absolute path to the project root, which is two directories up
22+ project_root = os .path .dirname (os .path .dirname (os .path .abspath (__file__ )))
23+ print (f"Project root: { project_root } " )
24+
25+ # Find the node executable in the system's PATH
26+ node_path = shutil .which ('node' )
27+ if not node_path :
28+ raise EnvironmentError ("Node.js executable not found. Please ensure Node.js is installed and in your PATH." )
29+
30+ # The command to start the server
31+ server_script = "dist/index.js"
32+ print (f"Running server command: { node_path } { server_script } " )
33+
34+ # Explicitly create an StdioTransport with the correct parameters
35+ transport = StdioTransport (command = node_path , args = [server_script ], cwd = project_root )
36+
37+ # The Client will use the specified transport
38+ async with Client (transport = transport ) as client :
39+ print ("MCP Client connected to server via STDIO." )
40+
41+ # Wait for the server to initialize its handlers
42+ print ("Waiting for server to initialize..." )
43+ await asyncio .sleep (2 )
44+
45+ # --- Confirm plugin connection ---
46+ print ("Waiting for Figma plugin to connect..." )
47+ plugin_connected = False
48+ retries = 0
49+ max_retries = 30 # Wait up to 30 * 1 second = 30 seconds
50+ while not plugin_connected and retries < max_retries :
51+ try :
52+ status_response = await client .call_tool ("figma_plugin_status" , arguments = {"operation" : "status" })
53+
54+ if status_response .content and status_response .content [0 ].text :
55+ status_data = yaml .safe_load (status_response .content [0 ].text )
56+ if status_data and status_data .get ("connected" ) == True :
57+ plugin_connected = True
58+ print ("Figma plugin connected!" )
59+ else :
60+ print (f"Plugin not yet connected. Retrying in 1 second... ({ retries + 1 } /{ max_retries } )" )
61+ await asyncio .sleep (1 )
62+ else :
63+ print (f"Could not get plugin status. Retrying in 1 second... ({ retries + 1 } /{ max_retries } )" )
64+ await asyncio .sleep (1 )
65+ except Exception as e :
66+ print (f"Error checking plugin status: { e } . Retrying in 1 second... ({ retries + 1 } /{ max_retries } )" )
67+ await asyncio .sleep (1 )
68+ retries += 1
69+
70+ if not plugin_connected :
71+ print ("Warning: Figma plugin did not connect within the expected time." )
72+
73+ # Get the list of tools
74+ tools = await client .list_tools ()
75+ tool_map = {tool .name : tool for tool in tools }
76+ tool_names = list (tool_map .keys ())
77+
78+ session = PromptSession ()
79+
80+ print ("\n Ready to receive commands." )
81+ print ("Type 'help' for a list of tools." )
82+ print ("Type 'help <tool_name>' for tool usage." )
83+ print ("Type 'tool_name, {arg1: value1, arg2: value2}' to call a tool." )
84+ print ("Type 'exit' to quit." )
85+
86+ while True :
87+ try :
88+ line = await session .prompt_async ("> " )
89+ line = line .strip ()
90+
91+ if not line :
92+ continue
93+
94+ if line .lower () == 'exit' :
95+ break
96+
97+ if line .lower () == 'help' :
98+ await _print_tool_list (tool_map )
99+ continue
100+
101+ if line .lower ().startswith ('help ' ):
102+ help_tool_name = line .split (' ' , 1 )[1 ]
103+ if help_tool_name in tool_map :
104+ tool_info = tool_map [help_tool_name ]
105+ print (f"\n Tool: { tool_info .name } " )
106+ print (f"Description: { tool_info .description } " )
107+ print ("Input Schema:" )
108+ print (yaml .dump (tool_info .inputSchema , indent = 2 ))
109+ else :
110+ print (f"Tool '{ help_tool_name } ' not found." )
111+ continue
112+
113+ try :
114+ # Parse the input as JSON string
115+ parts = line .split (', ' , 1 )
116+ if len (parts ) != 2 :
117+ print ("Invalid command format. Please use 'tool_name, {arg1: value1, ...}'." )
118+ continue
119+
120+ tool_name = parts [0 ].strip ()
121+ arguments_str = parts [1 ].strip ()
122+
123+ try :
124+ arguments = json .loads (arguments_str )
125+ except json .JSONDecodeError as e :
126+ print (f"Error parsing JSON arguments: { e } " )
127+ continue
128+
129+ if not isinstance (arguments , dict ):
130+ print ("Invalid arguments format. Arguments must be a JSON dictionary." )
131+ continue
132+
133+ if tool_name not in tool_map :
134+ print (f"Tool '{ tool_name } ' not found. Type 'help' for available tools." )
135+ continue
136+
137+ print (f"Calling tool '{ tool_name } ' with arguments: { arguments } " )
138+ response = await client .call_tool (tool_name , arguments = arguments )
139+ print ("Server response:" )
140+ if response .structured_content :
141+ print (yaml .dump (response .structured_content , indent = 2 ))
142+ elif response .content :
143+ print (response .content [0 ].text )
144+ else :
145+ print (response )
146+
147+ except Exception as e :
148+ print (f"Error executing command: { e } " )
149+
150+ except EOFError :
151+ break
152+ except KeyboardInterrupt :
153+ break
154+
155+ print ("Client session ended. Attempting to terminate MCP server..." )
156+
157+ if __name__ == "__main__" :
158+ asyncio .run (main ())
0 commit comments