good-mcp-client.py•3.2 kB
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import sys
class MCPClient:
def __init__(self):
# Initialize session and client objects
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
async def connect_to_server(self, server_script_path: str):
"""Connect to an MCP server
Args:
server_script_path: Path to the server script (.py or .js)
"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")
command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# List available tools
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
async def call_tool(self, tool_name, **kwargs):
result = await self.session.call_tool(tool_name, kwargs)
# Print the most informative representation of the result
if hasattr(result, 'content'):
print(f"\nResult: {result.content}")
elif hasattr(result, '__dict__'):
print(f"\nResult: {result.__dict__}")
else:
print(f"\nResult: {result}")
async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Client Started!")
print("Type 'insert' to insert a record, 'query' to query records, or 'quit' to exit.")
while True:
try:
cmd = input("\nCommand (insert/query/quit): ").strip().lower()
if cmd == 'insert':
name = input("Enter name: ")
address = input("Enter address: ")
await self.call_tool('insert_record', name=name, address=address)
elif cmd == 'query':
await self.call_tool('query_records')
elif cmd == 'quit':
break
else:
print("Unknown command.")
except Exception as e:
print(f"\nError: {str(e)}")
async def cleanup(self):
"""Clean up resources"""
await self.exit_stack.aclose()
async def main():
if len(sys.argv) < 2:
print("Usage: python good-mcp-client.py <path_to_server_script>")
sys.exit(1)
client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())