Computer Manager & Computer

from ufo.client.computer import ComputerManager
from ufo.client.mcp.mcp_server_manager import MCPServerManager
from config.config_loader import get_ufo_config

# 1. Get UFO configuration
ufo_config = get_ufo_config()

# 2. Initialize MCP server manager
mcp_server_manager = MCPServerManager()

# 3. Create computer manager
computer_manager = ComputerManager(
    ufo_config.to_dict(),
    mcp_server_manager
)

computer = Computer(
    name="default_agent",
    process_name="explorer.exe",
    mcp_server_manager=mcp_server_manager,
    data_collection_servers_config=[...],
    action_servers_config=[...]
)

# Async initialization (required)
await computer.async_init()

data_collection_servers:
  - namespace: screenshot_collector
    type: local
    module: ufo.client.mcp.local_servers.screenshot_server
    reset: false
  - namespace: ui_collector
    type: local
    module: ufo.client.mcp.local_servers.ui_server
    reset: false

action_servers:
  - namespace: gui_automator
    type: local
    module: ufo.client.mcp.local_servers.automation_server
    reset: false

async def execute(
    self,
    agent_name: str,
    process_name: str,
    root_name: str,
    commands: List[Command]
) -> List[Result]

def _call_tool_in_thread():
    """
    Execute MCP tool call in an isolated thread with its own event loop.
    This prevents blocking operations in MCP tools from blocking the main event loop.
    """
    # Create a new event loop for this thread
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        async def _do_call():
            async with Client(server) as client:
                return await client.call_tool(
                    name=tool_name, arguments=params, raise_on_error=False
                )
        return loop.run_until_complete(_do_call())
    finally:
        loop.close()

# Execute in thread pool with timeout protection
result = await asyncio.wait_for(
    loop.run_in_executor(self._executor, _call_tool_in_thread),
    timeout=self._tool_timeout
)

<tool_type>::<tool_name>

Examples:
- action::click
- action::type_text
- data_collection::screenshot
- data_collection::get_ui_elements

async def register_one_mcp_server(
    self, namespace: str, tool_type: str, mcp_server: BaseMCPServer
) -> None:
    async with Client(mcp_server.server) as client:
        tools = await client.list_tools()

        for tool in tools:
            tool_key = self.make_tool_key(tool_type, tool.name)

            self._register_tool(
                tool_key=tool_key,
                tool_name=tool.name,
                title=tool.title,
                namespace=namespace,
                tool_type=tool_type,
                description=tool.description,
                input_schema=tool.inputSchema,
                output_schema=tool.outputSchema,
                mcp_server=mcp_server
            )

@Computer.meta_tool("list_tools")
async def list_tools(
    self,
    tool_type: Optional[str] = None,
    namespace: Optional[str] = None,
    remove_meta: bool = True
) -> CallToolResult:
    """
    Get available tools of a specific type.
    """
    tools = []

    for tool in self._tools_registry.values():
        if ((tool_type is None or tool.tool_type == tool_type)
            and (namespace is None or tool.namespace == namespace)
            and (not remove_meta or tool.tool_name not in self._meta_tools)):
            tools.append(tool.tool_info.model_dump())

    return CallToolResult(
        content=[TextContent(type="text", text=json.dumps(tools))]
    )

# In __init__:
for attr in dir(self):
    method = getattr(self, attr)
    if callable(method) and hasattr(method, "_meta_tool_name"):
        name = getattr(method, "_meta_tool_name")
        self._meta_tools[name] = method

# Computer Manager reset (cascades to all computers)
computer_manager.reset()

# Computer instance reset
computer.reset()

import time
start = time.time()
result = await computer._run_action(tool_call)
duration = time.time() - start
if duration > 300:  # 5 minutes
    logger.warning(f"Slow tool: {tool_call.tool_name} took {duration}s")

# 100-minute timeout is generous but not infinite
# Design tools to complete within reasonable time

# Separate data collection from actions
data_tools = await computer.list_tools(tool_type="data_collection")
action_tools = await computer.list_tools(tool_type="action")