Spaces:
Running
Running
File size: 2,383 Bytes
e074e5b d4c3cf0 e074e5b d4c3cf0 e074e5b d4c3cf0 e074e5b d4c3cf0 e074e5b d4c3cf0 e074e5b ea31a57 d4c3cf0 e074e5b ea31a57 d4c3cf0 e074e5b d4c3cf0 e074e5b ea31a57 d4c3cf0 e074e5b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from time import sleep
import base64
import io
from PIL import Image
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
org_or_user = "a-mahla"
space_name = "test_desktop"
@asynccontextmanager
async def get_mcp_session() -> AsyncGenerator[ClientSession, None]:
"""Context manager for recording the demo with MCP server"""
base_url = f"https://{org_or_user}-{space_name}.hf.space".replace("_", "-")
server_url = f"{base_url}/mcp/"
print("🎬 MCP Server started:", server_url)
# Connect to the server
async with streamablehttp_client(server_url) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
yield session
async def main():
async with get_mcp_session() as session:
response = await session.list_tools()
print("Available Tools:")
print("=" * 50)
for tool in response.tools:
print("-" * 50)
print(f"📋 '{tool.name}':", end="")
if tool.description:
desc = tool.description.strip()
print(f" {desc}", end="\n")
else:
print(" No description", end="\n")
print()
print("Opening https://www.huggingface.co for research...")
await session.call_tool("open", {"file_or_url": "https://www.huggingface.co/"})
sleep(7)
print(await session.call_tool("move_mouse", {"x": 1200, "y": 120}))
sleep(2)
print(await session.call_tool("left_click", {}))
sleep(2)
print(await session.call_tool("move_mouse", {"x": 1200, "y": 160}))
print(await session.call_tool("left_click", {}))
sleep(2)
print(await session.call_tool("move_mouse", {"x": 1600, "y": 320}))
print(await session.call_tool("left_click", {}))
sleep(2)
response = await session.call_tool("screenshot", {})
screenshot_base64 = response.content[0].data
screenshot_bytes = base64.b64decode(screenshot_base64)
image = Image.open(io.BytesIO(screenshot_bytes))
image.save("screenshot.png")
if __name__ == "__main__":
asyncio.run(main()) |