Reducing Code in Python AgentsFramework Sample

python/agent-framework/sample-agent/agent.py

@@ -38,29 +38,28 @@
 # AgentFramework SDK
 from agent_framework import ChatAgent
 from agent_framework.azure import AzureOpenAIChatClient
-from agent_framework.observability import setup_observability
 
 # Agent Interface
 from agent_interface import AgentInterface
 from azure.identity import AzureCliCredential
 
 # Microsoft Agents SDK
-from local_authentication_options import LocalAuthenticationOptions
 from microsoft_agents.hosting.core import Authorization, TurnContext
 
-# Observability Components
-from microsoft_agents_a365.observability.extensions.agentframework.trace_instrumentor import (
-    AgentFrameworkInstrumentor,
-)
-
 # MCP Tooling
 from microsoft_agents_a365.tooling.extensions.agentframework.services.mcp_tool_registration_service import (
     McpToolRegistrationService,
 )
-from token_cache import get_cached_agentic_token
+
+# Observability
+from microsoft_agents_a365.observability.extensions.agentframework.trace_instrumentor import (
+    AgentFrameworkInstrumentor,
+)
 
 # </DependencyImports>
 
+class LocalAuthenticationOptions():
+    bearer_token: str
 
 class AgentFrameworkAgent(AgentInterface):
     """AgentFramework Agent integrated with MCP servers and Observability"""
@@ -74,21 +73,19 @@ def __init__(self):
         """Initialize the AgentFramework agent."""
         self.logger = logging.getLogger(self.__class__.__name__)
 
-        # Initialize auto instrumentation with Agent365 observability SDK
-        self._enable_agentframework_instrumentation()
+        # Initialize Agent 365 Observability Wrapper for AgentFramework SDK
+        AgentFrameworkInstrumentor().instrument()
 
         # Initialize authentication options
-        self.auth_options = LocalAuthenticationOptions.from_environment()
+        self.auth_options = LocalAuthenticationOptions()
+        self.auth_options.bearer_token = os.getenv("BEARER_TOKEN", "")
+
+        # Initialize MCP services
+        self.tool_service = McpToolRegistrationService()
 
         # Create Azure OpenAI chat client
         self._create_chat_client()
 
-        # Create the agent with initial configuration
-        self._create_agent()
-
-        # Initialize MCP services
-        self._initialize_services()
-
     # </Initialization>
 
     # =========================================================================
@@ -122,121 +119,47 @@ def _create_chat_client(self):
             api_version=api_version,
         )
 
-        logger.info("✅ AzureOpenAIChatClient created successfully")
+        logger.info("AzureOpenAIChatClient created successfully")
 
     def _create_agent(self):
         """Create the AgentFramework agent with initial configuration"""
         try:
-            logger.info("Creating AgentFramework agent...")
-
             self.agent = ChatAgent(
                 chat_client=self.chat_client,
                 instructions="You are a helpful assistant with access to tools.",
                 tools=[],  # Tools will be added dynamically by MCP setup
             )
 
-            logger.info("✅ AgentFramework agent created successfully")
-
         except Exception as e:
             logger.error(f"Failed to create agent: {e}")
             raise
 
     # </ClientCreation>
 
     # =========================================================================
-    # OBSERVABILITY CONFIGURATION
-    # =========================================================================
-    # <ObservabilityConfiguration>
-
-    def token_resolver(self, agent_id: str, tenant_id: str) -> str | None:
-        """
-        Token resolver function for Agent 365 Observability exporter.
-
-        Uses the cached agentic token obtained from AGENT_APP.auth.get_token(context, "AGENTIC").
-        This is the only valid authentication method for this context.
-        """
-
-        try:
-            logger.info(
-                f"Token resolver called for agent_id: {agent_id}, tenant_id: {tenant_id}"
-            )
-
-            # Use cached agentic token from agent authentication
-            cached_token = get_cached_agentic_token(tenant_id, agent_id)
-            if cached_token:
-                logger.info("Using cached agentic token from agent authentication")
-                return cached_token
-            else:
-                logger.warning(
-                    f"No cached agentic token found for agent_id: {agent_id}, tenant_id: {tenant_id}"
-                )
-                return None
-
-        except Exception as e:
-            logger.error(
-                f"Error resolving token for agent {agent_id}, tenant {tenant_id}: {e}"
-            )
-            return None
-
-    def _setup_observability(self):
-        """
-        Configure observability using agent_framework.observability.setup_observability()
-        """
-        try:
-            setup_observability()
-            logger.info("✅ AgentFramework observability configured successfully")
-        except Exception as e:
-            logger.error(f"❌ Error setting up observability: {e}")
-
-    def _enable_agentframework_instrumentation(self):
-        """Enable AgentFramework instrumentation for automatic tracing"""
-        try:
-            # Initialize Agent 365 Observability Wrapper for AgentFramework SDK
-            AgentFrameworkInstrumentor().instrument()
-            logger.info("✅ AgentFramework instrumentation enabled")
-        except Exception as e:
-            logger.warning(f"⚠️ Could not enable AgentFramework instrumentation: {e}")
-
-    # </ObservabilityConfiguration>
-
-    # =========================================================================
-    # MCP SERVER SETUP AND INITIALIZATION
+    # MCP SERVER SETUP
     # =========================================================================
     # <McpServerSetup>
 
-    def _initialize_services(self):
-        """Initialize MCP services and authentication options"""
-        try:
-            # Create MCP tool registration service
-            self.tool_service = McpToolRegistrationService()
-            logger.info("✅ AgentFramework MCP tool registration service initialized")
-        except Exception as e:
-            logger.warning(f"⚠️ Could not initialize MCP tool service: {e}")
-            self.tool_service = None
-
-    async def setup_mcp_servers(self, auth: Authorization, context: TurnContext):
+    async def _create_agent_with_mcp(self, auth: Authorization, context: TurnContext):
         """Set up MCP server connections"""
         try:
             if not self.tool_service:
                 logger.warning(
-                    "⚠️ MCP tool service not available -  skipping MCP server setup"
+                    "MCP tool service not available -  skipping MCP server setup"
                 )
                 return
 
-            logger.info("🔍 Starting MCP server setup...")
+            logger.info("Starting MCP server setup...")
 
             agent_user_id = os.getenv("AGENT_ID", "user123")
             use_agentic_auth = os.getenv("USE_AGENTIC_AUTH", "false").lower() == "true"
 
-            logger.info(f"🆔 Agent User ID: {agent_user_id}")
-            logger.info(f"🔐 Using agentic auth: {use_agentic_auth}")
-
             if use_agentic_auth:
-                logger.info("🔄 Adding tool servers with agentic authentication...")
                 scope = os.getenv("AGENTIC_AUTH_SCOPE")
                 if not scope:
                     logger.warning(
-                        "⚠️ AGENTIC_AUTH_SCOPE environment variable is not set when USE_AGENTIC_AUTH=true"
+                        "AGENTIC_AUTH_SCOPE environment variable is not set when USE_AGENTIC_AUTH=true"
                     )
                     return
                 scopes = [scope]
@@ -247,58 +170,44 @@ async def setup_mcp_servers(self, auth: Authorization, context: TurnContext):
                     agent_instructions="You are a helpful assistant with access to tools.",
                     initial_tools=[],
                     agentic_app_id=agent_user_id,
-                    environment_id=self.auth_options.env_id,
+                    environment_id="",
                     auth=auth,
                     turn_context=context,
                     auth_token=auth_token,
                 )
             else:
-                logger.info(
-                    "🔄 Adding tool servers with bearer token authentication..."
-                )
                 self.agent = await self.tool_service.add_tool_servers_to_agent(
                     chat_client=self.chat_client,
                     agent_instructions="You are a helpful assistant with access to tools.",
                     initial_tools=[],
                     agentic_app_id=agent_user_id,
-                    environment_id=self.auth_options.env_id,
+                    environment_id="",
                     auth=auth,
                     auth_token=self.auth_options.bearer_token,
                     turn_context=context,
                 )
-
-            if self.agent:
-                logger.info("✅ Agent MCP setup completed successfully")
-            else:
-                logger.warning("⚠️ Agent MCP setup returned None")
-
         except Exception as e:
             logger.error(f"Error setting up MCP servers: {e}")
             logger.exception("Full error details:")
 
+        if not self.agent:
+            logger.warning("Agent MCP setup returned None, returning agent without servers.")
+            self._create_agent()
+
     # </McpServerSetup>
 
     # =========================================================================
-    # INITIALIZATION AND MESSAGE PROCESSING
+    # MESSAGE PROCESSING
     # =========================================================================
     # <MessageProcessing>
 
-    async def initialize(self):
-        """Initialize the agent and MCP server connections"""
-        logger.info("Initializing AgentFramework Agent with MCP servers...")
-        try:
-            logger.info("Agent initialized successfully")
-        except Exception as e:
-            logger.error(f"Failed to initialize agent: {e}")
-            raise
-
     async def process_user_message(
         self, message: str, auth: Authorization, context: TurnContext
     ) -> str:
         """Process user message using the AgentFramework SDK"""
         try:
             # Setup MCP servers
-            await self.setup_mcp_servers(auth, context)
+            await self._create_agent_with_mcp(auth, context)
 
             # Run the agent with the user message
             result = await self.agent.run(message)
@@ -330,48 +239,13 @@ async def process_user_message(
     async def cleanup(self) -> None:
         """Clean up agent resources and MCP server connections"""
         try:
-            logger.info("Cleaning up agent resources...")
-
             # Cleanup MCP tool service if it exists
             if hasattr(self, "tool_service") and self.tool_service:
                 try:
                     await self.tool_service.cleanup()
-                    logger.info("MCP tool service cleanup completed")
                 except Exception as cleanup_ex:
                     logger.warning(f"Error cleaning up MCP tool service: {cleanup_ex}")
-
-            logger.info("Agent cleanup completed")
-
         except Exception as e:
             logger.error(f"Error during cleanup: {e}")
 
-    # </Cleanup>
-
-
-# =============================================================================
-# MAIN ENTRY POINT
-# =============================================================================
-# <MainEntryPoint>
-
-
-async def main():
-    """Main function to run the AgentFramework Agent with MCP servers"""
-    try:
-        # Create and initialize the agent
-        agent = AgentFrameworkAgent()
-        await agent.initialize()
-
-    except Exception as e:
-        logger.error(f"Failed to start agent: {e}")
-        print(f"Error: {e}")
-
-    finally:
-        # Cleanup
-        if "agent" in locals():
-            await agent.cleanup()
-
-
-if __name__ == "__main__":
-    asyncio.run(main())
-
-# </MainEntryPoint>
+    # </Cleanup>

python/agent-framework/sample-agent/agent_interface.py

@@ -8,20 +8,13 @@
 from abc import ABC, abstractmethod
 from microsoft_agents.hosting.core import Authorization, TurnContext
 
-
 class AgentInterface(ABC):
     """
     Abstract base class that any hosted agent must inherit from.
 
     This ensures agents implement the required methods at class definition time,
     providing stronger guarantees than a Protocol.
     """
-
-    @abstractmethod
-    async def initialize(self) -> None:
-        """Initialize the agent and any required resources."""
-        pass
-
     @abstractmethod
     async def process_user_message(
         self, message: str, auth: Authorization, context: TurnContext
@@ -32,22 +25,4 @@ async def process_user_message(
     @abstractmethod
     async def cleanup(self) -> None:
         """Clean up any resources used by the agent."""
-        pass
-
-
-def check_agent_inheritance(agent_class) -> bool:
-    """
-    Check that an agent class inherits from AgentInterface.
-
-    Args:
-        agent_class: The agent class to check
-
-    Returns:
-        True if the agent inherits from AgentInterface, False otherwise
-    """
-    if not issubclass(agent_class, AgentInterface):
-        print(f"❌ Agent {agent_class.__name__} does not inherit from AgentInterface")
-        return False
-
-    print(f"✅ Agent {agent_class.__name__} properly inherits from AgentInterface")
-    return True
+        pass