1+ # Copyright (c) Microsoft. All rights reserved.
2+
3+ import os
4+ import asyncio
5+ from azure .identity .aio import DefaultAzureCredential
6+ from azure .identity import InteractiveBrowserCredential
7+
8+ from azure .ai .agents .models import BingGroundingTool
9+ from azure .ai .agents .models import CodeInterpreterToolDefinition
10+ from semantic_kernel .connectors .mcp import MCPStreamableHttpPlugin
11+ from semantic_kernel .agents .azure_ai .azure_ai_agent import AzureAIAgent
12+ from semantic_kernel .agents .azure_ai .azure_ai_agent_settings import \
13+ AzureAIAgentSettings
14+
15+ # Environment variables are used in two ways:
16+ # 1. Automatically by AzureAIAgentSettings (AZURE_OPENAI_... variables)
17+ # 2. Directly created below (for bing grounding and MCP capabilities)
18+
19+ class FoundryAgentTemplate :
20+ """A template agent that manages its own async context for client connections."""
21+
22+ AGENT_NAME = "CoderAgent"
23+ AGENT_DESCRIPTION = "A coding assistant with Azure AI capabilities and code execution tools"
24+ AGENT_INSTRUCTIONS = """You solve questions using code. Please provide detailed analysis and
25+ computation process. You work with data provided by other agents in the team."""
26+
27+ initialized = False
28+
29+ # To do: pass capability parameters in the constructor:
30+ # To do: pass name, description and instructions in the constructor
31+ # 1. MCP server endpoint and use
32+ # 2. Bing grounding option
33+ # 3. Reasoning model name (some settings are different and cannot be used with bing grounding)
34+ # 4. Coding skills - CodeInterpreterToolDefinition
35+ # 5. Grounding Data - requires index endpoint
36+ # This will allow the factory to create all base models except researcher with bing - this is
37+ # is coming with a deep research offering soon (preview now in two regions)
38+ def __init__ (self ):
39+ self ._agent = None
40+ self .client = None
41+ self .creds = None
42+ self .mcp_plugin = None
43+ self .bing_tool_name = os .environ ["BING_CONNECTION_NAME" ] or ""
44+ self .mcp_srv_endpoint = os .environ ["MCP_SERVER_ENDPOINT" ] or ""
45+ self .mcp_srv_name = os .environ ["MCP_SERVER_NAME" ] or ""
46+ self .mcp_srv_description = os .environ ["MCP_SERVER_DESCRIPTION" ] or ""
47+ self .tenant_id = os .environ ["TENANT_ID" ] or ""
48+ self .client_id = os .environ ["CLIENT_ID" ] or ""
49+
50+ def __getattr__ (self , name ):
51+ """Delegate all attribute access to the wrapped agent."""
52+ if hasattr (self , '_agent' ) and self ._agent is not None :
53+ return getattr (self ._agent , name )
54+ else :
55+ raise AttributeError (f"'{ type (self ).__name__ } { name } )
56+
57+ async def __aenter__ (self ):
58+ """Initialize the agent and return it within an async context."""
59+ # Initialize credentials and client
60+ if not self .initialized :
61+ await self .create_agent_async ()
62+ return self
63+
64+ async def __aexit__ (self , exc_type , exc_val , exc_tb ):
65+ """Clean up the async contexts."""
66+ # Exit MCP plugin context first
67+ if self .initialized :
68+ await self .close ()
69+
70+ async def create_agent_async (self ):
71+ """Create the template agent with all tools - must be called within async context."""
72+ self .initialized = True
73+
74+ self .creds = DefaultAzureCredential ()
75+ self .client = AzureAIAgent .create_client (credential = self .creds )
76+
77+ # Get MCP authentication headers
78+ headers = await self ._get_mcp_auth_headers ()
79+
80+ # Create Bing tools
81+ bing = None
82+ try :
83+ bing_connection = await self .client .connections .get (name = self .bing_tool_name )
84+ conn_id = bing_connection .id
85+ print (f"🔍 Attempting Bing tool creation with connection name: { self .bing_tool_name } )
86+ bing = BingGroundingTool (connection_id = conn_id )
87+ print (f"🔍 Bing tool created with { conn_id } { len (bing .definitions )} )
88+ except Exception as name_error :
89+ print (f"⚠️ Bing tool creation with { self .bing_tool_name } { name_error } )
90+
91+ # Create MCP plugin and enter its async context
92+ try :
93+ print ("🔗 Creating MCP plugin within async context..." )
94+ self .mcp_plugin = MCPStreamableHttpPlugin (
95+ name = self .mcp_srv_name ,
96+ description = self .mcp_srv_description ,
97+ url = self .mcp_srv_endpoint ,
98+ headers = headers ,
99+ )
100+
101+ # Enter the MCP plugin's async context
102+ if hasattr (self .mcp_plugin , '__aenter__' ):
103+ await self .mcp_plugin .__aenter__ ()
104+ print ("✅ MCP plugin async context entered" )
105+ else :
106+ print ("ℹ️ MCP plugin doesn't require async context" )
107+
108+ except Exception as mcp_error :
109+ print (f"⚠️ MCP plugin creation failed: { mcp_error } )
110+ self .mcp_plugin = None
111+
112+ # Create agent settings and definition
113+ ai_agent_settings = AzureAIAgentSettings ()
114+ template_agent_definition = await self .client .agents .create_agent (
115+ model = ai_agent_settings .model_deployment_name ,
116+ # Name, description and instructions are provided for demonstration purposes
117+ name = self .AGENT_NAME ,
118+ description = self .AGENT_DESCRIPTION ,
119+ instructions = self .AGENT_INSTRUCTIONS ,
120+ # tools=bing.definitions if bing else [],
121+ # Add Code Interpreter tool for coding capabilities
122+ tools = [CodeInterpreterToolDefinition ()]
123+ )
124+
125+ # Create the final agent
126+ plugins = [self .mcp_plugin ] if self .mcp_plugin else []
127+ self ._agent = AzureAIAgent (
128+ client = self .client ,
129+ definition = template_agent_definition ,
130+ plugins = plugins
131+ )
132+
133+ print ("✅ Template agent created successfully!" )
134+
135+ async def close (self ):
136+ """Clean up async resources."""
137+ if not self .initialized :
138+ return
139+ # Exit MCP plugin context first
140+ if self .mcp_plugin :
141+ #await self.mcp_plugin.__aexit__(None, None, None)
142+ self .mcp_plugin = None
143+ try :
144+ # Then exit Azure contexts
145+ if self .client :
146+ await self .client .__aexit__ (None , None , None )
147+ except Exception as e :
148+ print (f"⚠️ { self .AGENT_NAME } { e } )
149+ try :
150+ if self .creds :
151+ await self .creds .__aexit__ (None , None , None )
152+ except Exception as e :
153+ print (f"⚠️ { self .AGENT_NAME } { e } )
154+
155+ self .initialized = False
156+
157+ # Add __del__ for emergency cleanup
158+ def __del__ (self ):
159+ """Emergency cleanup when object is garbage collected."""
160+ if self .initialized :
161+ try :
162+ # Try to schedule cleanup in the event loop
163+ import asyncio
164+ loop = asyncio .get_event_loop ()
165+ if loop .is_running ():
166+ loop .create_task (self .close ())
167+ except Exception :
168+ # If we can't schedule cleanup, just warn
169+ print (f"⚠️ Warning: { self .AGENT_NAME } )
170+
171+ async def _get_mcp_auth_headers (self ) -> dict :
172+ """Get MCP authentication headers."""
173+ try :
174+ interactive_credential = InteractiveBrowserCredential (
175+ tenant_id = self .tenant_id ,
176+ client_id = self .client_id
177+ )
178+ token = interactive_credential .get_token (f"api://{ self .client_id } )
179+ headers = {
180+ "Authorization" : f"Bearer { token .token } ,
181+ "Content-Type" : "application/json"
182+ }
183+ print ("✅ Successfully obtained MCP authentication token" )
184+ return headers
185+ except Exception as e :
186+ print (f"❌ Failed to get MCP token: { e } )
187+ return {}
188+
189+
190+ # Factory function for your agent factory
191+ # Add parameters to allow creation of agents with different capabilities
192+ async def create_foundry_agent ():
193+ """Factory function that returns a AzureAiAgentTemplate context manager."""
194+ return_agent = FoundryAgentTemplate ()
195+ await return_agent .create_agent_async ()
196+ return return_agent
197+
198+
199+ # Test harness
200+ async def test_agent ():
201+ """Simple chat test harness for the agent."""
202+ print ("🤖 Starting agent test harness..." )
203+
204+ try :
205+ async with FoundryAgentTemplate () as agent :
206+ print ("💬 Type 'quit' or 'exit' to stop\n " )
207+
208+ while True :
209+ user_input = input ("You: " ).strip ()
210+
211+ if user_input .lower () in ['quit' , 'exit' , 'q' ]:
212+ print ("👋 Goodbye!" )
213+ break
214+
215+ if not user_input :
216+ continue
217+
218+ try :
219+ print ("🤖 Agent: " , end = "" , flush = True )
220+ async for message in agent .invoke (user_input ):
221+ if hasattr (message , 'content' ):
222+ print (message .content , end = "" , flush = True )
223+ else :
224+ print (str (message ), end = "" , flush = True )
225+ print ()
226+
227+ except Exception as e :
228+ print (f"❌ Error: { e } )
229+
230+ except Exception as e :
231+ print (f"❌ Failed to create agent: { e } )
232+
233+
234+ if __name__ == "__main__" :
235+ asyncio .run (test_agent ())
0 commit comments