count_files_script.py

import asyncio
import json
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # 创建服务器参数（使用标准IO连接）
    server_params = StdioServerParameters(
        command="uv", 
        args=["--directory", ".", "run", "python", "file_server.py"]
    )
    
    path_to_count = "/Users/littlefatz/workspace/mcp-demo111"
    
    # 连接到服务器
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            
            # 调用count_files工具
            print(f"正在统计路径 '{path_to_count}' 下的文件数量...")
            tool_result = await session.call_tool(
                "count_files", 
                arguments={"directory_path": path_to_count}
            )
            
            # 从工具结果中提取JSON内容
            if hasattr(tool_result, 'content') and tool_result.content:
                for content_item in tool_result.content:
                    if hasattr(content_item, 'text'):
                        try:
                            result = json.loads(content_item.text)
                            print(f"文件总数: {result.get('file_count', '未知')}")
                            print(f"目录总数: {result.get('directory_count', '未知')}")
                            
                            if 'file_extensions' in result:
                                print("\n文件类型统计:")
                                extensions = result['file_extensions']
                                for ext, count in extensions.items():
                                    ext_display = ext if ext else '无扩展名'
                                    print(f"  {ext_display}: {count}个文件")
                        except json.JSONDecodeError:
                            print(f"无法解析JSON: {content_item.text}")
            else:
                print(f"未能获取有效结果: {tool_result}")

if __name__ == "__main__":
    asyncio.run(main())