通过MCP一键集成和Computer Use Agent模型，实现Azure Foundry中的AI自动化

 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

async def main() -> str:
    conversation_history = []
    generated_reports = []
    
    while True:
        user_query = input("Enter your query for fashion trends:->  ")
        new_user_message = {
            "role": "user",
            "content": [{"type": "input_text", "text": user_query}],
        }
        conversation_history.append(new_user_message)
        
        response = ai_client.create_app_response(
            instructions=instructions,
            conversation_history=conversation_history,
            mcp_server_url=config.mcp_server_url,
            available_functions=available_functions,
        )
        
        for output in response.output:
            if output.type == "function_call":
                function_to_call = available_functions[output.name]
                function_args = json.loads(output.arguments)
                function_response = await function_to_call(**function_args)
            elif output.type == "mcp_tool_call":
                print(f"MCP tool call: {output.name}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

class TrendsCompiler:
    async def compile_trends(self, user_query: str) -> str:
        async with LocalPlaywrightComputer() as computer:
            state = {"trends_compiled": False}
            step = 0
            while not state["trends_compiled"]:
                try:
                    if step == 0:
                        await self._launch_pinterest(computer)
                        step += 1
                    elif step == 1:
                        coordinates = await self._search_and_get_coordinates(computer, user_query)
                        if coordinates:
                            step += 1
                    elif step == 2:
                        await self._process_image_results(computer, coordinates, user_query)
                        markdown_report = await self._generate_markdown_report(user_query)
                        state["trends_compiled"] = True
                except Exception as e:
                    print(f"Autonomous error handling in step {step}: {e}")
                    state["trends_compiled"] = True
            return markdown_report

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

async def _detect_search_results(self, computer) -> List[Tuple[int, int, int, int]]:
    screenshot_bytes = await computer.screenshot()
    screenshot_b64 = base64.b64encode(screenshot_bytes).decode()
    prompt = """
    Analyze this Pinterest search results page and identify all trend/fashion 
    images displayed. For each image, provide the exact bounding box coordinates 
    in the format: <click>x1,y1,x2,y2</click>
    Focus on the main content images, not navigation or advertisement elements.
    """
    response = await self.ai_client.create_cua_response(
        prompt=prompt,
        screenshot_b64=screenshot_b64
    )
    coordinates = self.coordinate_parser.extract_coordinates(response.content)
    print(f"CUA model identified {len(coordinates)} image regions")
    return coordinates

1
2
3
4
5
6
7

def calculate_centers(self, coordinates: List[Tuple[int, int, int, int]]) -> List[Tuple[int, int]]:
    centers = []
    for x1, y1, x2, y2 in coordinates:
        center_x = (x1 + x2) // 2
        center_y = (y1 + y2) // 2
        centers.append((center_x, center_y))
    return centers

 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

async def _generate_markdown_report(self, user_query: str) -> str:
    if not self.image_analyses:
        return "No trend data collected for analysis."
    report_sections = [
        f"# Fashion Trends Analysis: {user_query}",
        f"*Generated on {datetime.now().strftime('%B %d, %Y')}*",
        "",
        "## Executive Summary",
        await self._generate_executive_summary(),
        "",
        "## Detailed Trend Analysis"
    ]
    for idx, analysis in enumerate(self.image_analyses, 1):
        trend_section = [
            f"### Trend Item {idx}",
            analysis.get('analysis', 'No analysis available'),
            f"*Analysis timestamp: {analysis.get('timestamp', 'Unknown')}*",
            ""
        ]
        report_sections.extend(trend_section)
    report_sections.extend([
        "## Trend Synthesis and Insights",
        await self._generate_trend_synthesis(),
        "",
        "## Recommendations",
        await self._generate_recommendations()
    ])
    return "\n".join(report_sections)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def create_app_tools(self, mcp_server_url: str, available_functions: Dict[str, Any]) -> List[Dict[str, Any]]:
    tools = [
        {
            "type": "mcp",
            "server_label": "azure-storage-mcp-server",
            "server_url": mcp_server_url,
            "require_approval": "never",
            "allowed_tools": ["create_container", "list_containers", "upload_blob"],
        }
    ]
    return tools