diff --git a/.gitignore b/.gitignore index 2dde7a18..938dae64 100644 --- a/.gitignore +++ b/.gitignore @@ -43,6 +43,8 @@ ScaffoldingReadMe.txt ~$* *~ CodeCoverage/ +driving_safety_report.md + # MSBuild Binary and Structured Log *. diff --git a/python/crewai/sample_agent/.env.template b/python/crewai/sample_agent/.env.template index e930d151..897a50fc 100644 --- a/python/crewai/sample_agent/.env.template +++ b/python/crewai/sample_agent/.env.template @@ -1,54 +1,112 @@ -# Demo environment for CrewAI hosted agent +# ============================================================================= +# CrewAI Agent Sample - Environment Configuration +# ============================================================================= +# Copy this file to .env and fill in your values. +# Lines starting with # are comments. Remove # to enable a variable. +# ============================================================================= + +# ----------------------------------------------------------------------------- +# OPENAI / AZURE OPENAI CONFIGURATION +# ----------------------------------------------------------------------------- +# Choose ONE of the following configurations: + +# Option A: Standard OpenAI +# Get your API key from https://platform.openai.com/api-keys OPENAI_API_KEY= -# MCP Server Configuration -MCP_SERVER_PORT=8000 -MCP_SERVER_HOST=localhost -MCP_DEVELOPMENT_BASE_URL= +# Option B: Azure OpenAI (recommended for enterprise) +# Get these values from Azure Portal > Your OpenAI Resource > Keys and Endpoint +AZURE_API_KEY= +AZURE_API_BASE= +AZURE_API_VERSION=2025-01-01-preview + +# Model Configuration +# For Azure OpenAI: Use "azure/" format (e.g., azure/gpt-4.1) +# For OpenAI: Use model name directly (e.g., gpt-4o-mini) +OPENAI_MODEL_NAME=azure/gpt-4.1 + +# ----------------------------------------------------------------------------- +# MCP (MODEL CONTEXT PROTOCOL) AUTHENTICATION +# ----------------------------------------------------------------------------- +# These settings enable MCP tools like Mail, Calendar, and Copilot + +# Bearer Token for MCP Server Authentication +# Generate with: a365 develop get-token -o raw +# This token expires - regenerate when you see MCP connection errors +BEARER_TOKEN= -# Logging -LOG_LEVEL=INFO +# Agentic Authentication Mode +# Set to "false" for local development with bearer token +# Set to "true" for production with full app registration +USE_AGENTIC_AUTH=false -# Observability Configuration -OBSERVABILITY_SERVICE_NAME=crewai-agent-sample +# Agent identifiers +# AGENT_ID is the primary identifier used by the backend for observability. +# If not set, the application automatically falls back to using AGENTIC_APP_ID. +AGENT_ID= + +# Agent Application ID (used for MCP tool discovery and as the default agent ID) +# This identifies your agent when connecting to MCP servers +AGENTIC_APP_ID=crewai-agent +# ----------------------------------------------------------------------------- +# TAVILY API - WEATHER SEARCH TOOL +# ----------------------------------------------------------------------------- +# Required for the WeatherTool to search current weather conditions +# Get your free API key from https://tavily.com +TAVILY_API_KEY= + +# ----------------------------------------------------------------------------- +# OBSERVABILITY CONFIGURATION +# ----------------------------------------------------------------------------- +# These settings configure Agent 365 observability tracing + +# Service identifiers for telemetry +OBSERVABILITY_SERVICE_NAME=crewai-agent-sample OBSERVABILITY_SERVICE_NAMESPACE=agent365-samples +# Enable/disable observability features +ENABLE_OBSERVABILITY=true -BEARER_TOKEN= +# Enable Agent 365 cloud exporter (requires valid token) +# Set to "true" to send traces to Agent 365 observability backend +ENABLE_A365_OBSERVABILITY_EXPORTER=false -OPENAI_MODEL=gpt-4o-mini +# Python environment indicator +PYTHON_ENVIRONMENT=development -USE_AGENTIC_AUTH=false +# ----------------------------------------------------------------------------- +# SERVER CONFIGURATION +# ----------------------------------------------------------------------------- +# The port the agent server listens on +# If busy, the server will automatically try the next available port +PORT=3978 -AGENTIC_APP_ID= +# Logging verbosity: DEBUG, INFO, WARNING, ERROR +LOG_LEVEL=INFO -# Agent observability attributes (optional but recommended) -# These populate the gen_ai.agent.* attributes in traces -AGENT_AUID=your_agent_auid +# ----------------------------------------------------------------------------- +# MCP SERVER CONFIGURATION (ADVANCED) +# ----------------------------------------------------------------------------- +# These are typically not needed for standard usage -# Agent ID (required for agentic authentication) -AGENT_ID=your-agent-id +# Local MCP server settings (for custom MCP server development) +MCP_SERVER_PORT=8000 +MCP_SERVER_HOST=localhost +MCP_DEVELOPMENT_BASE_URL= -# Agent 365 Agentic Authentication Configuration -# Auth handler name (optional, defaults to "AGENTIC") -# The name of the authentication handler to use for token exchange -AGENT_AUTH_HANDLER_NAME=AGENTIC +# ----------------------------------------------------------------------------- +# AGENT 365 APP REGISTRATION (PRODUCTION) +# ----------------------------------------------------------------------------- +# Required for production deployments with full authentication +# Get these from Azure Portal > App Registrations > Your App +# Azure AD App Registration CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID= CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET= CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID= CONNECTIONS__SERVICE_CONNECTION__SETTINGS__SCOPES=https://graph.microsoft.com/.default +# Service URL mapping CONNECTIONSMAP__0__SERVICEURL=* CONNECTIONSMAP__0__CONNECTION=SERVICE_CONNECTION - -# Optional: Server Configuration -PORT=3978 - -# Required for observability SDK -ENABLE_OBSERVABILITY=true - -ENABLE_A365_OBSERVABILITY_EXPORTER=false - -PYTHON_ENVIRONMENT=development diff --git a/python/crewai/sample_agent/README.md b/python/crewai/sample_agent/README.md index 13986d99..85679464 100644 --- a/python/crewai/sample_agent/README.md +++ b/python/crewai/sample_agent/README.md @@ -1,10 +1,14 @@ -# CrewAgent Crew - Python +# CrewAI Agent Sample - Python -This sample demonstrates how to build a multi‑agent system using CrewAI while integrating with the Microsoft Agent 365 SDK. It mirrors the structure and hosting patterns of the AgentFramework/OpenAI Agent 365 samples, while preserving native CrewAI logic in src/crew_agent. It covers: +This sample demonstrates how to build a multi-agent system using CrewAI while integrating with the Microsoft Agent 365 SDK. It mirrors the structure and hosting patterns of the AgentFramework/OpenAI Agent 365 samples, while preserving native CrewAI logic in `src/crew_agent/`. -- **Observability**: End-to-end tracing, caching, and monitoring for agent applications -- **Notifications**: Services and models for managing user notifications -- **Tools**: Model Context Protocol tools for building advanced agent solutions +## Demonstrates + +- **MCP Tooling**: Full Model Context Protocol (MCP) integration with Microsoft 365 services (Mail, Calendar, Copilot) +- **Observability**: Complete tracing with `InvokeAgentScope`, `InferenceScope`, and `ExecuteToolScope` for all agent operations +- **Notifications**: Email and Teams @mention notification support +- **Multi-Agent Orchestration**: Sequential agent workflow with Weather Checker and Driving Safety Advisor agents +- **Azure OpenAI Integration**: Native support for Azure OpenAI deployments via CrewAI's `azure-ai-inference` extra and LiteLLM routing - **Hosting Patterns**: Hosting with Microsoft 365 Agents SDK This sample uses the [Microsoft Agent 365 SDK for Python](https://github.com/microsoft/Agent365-python). @@ -14,48 +18,168 @@ For comprehensive documentation and guidance on building agents with the Microso ## Prerequisites - Python 3.11+ -- Microsoft Agent 365 SDK -- Azure/OpenAI API credentials -- UV (recommended for dependency management) +- [UV](https://docs.astral.sh/uv/) (recommended for dependency management) +- Azure OpenAI API credentials OR OpenAI API key +- [Tavily API key](https://tavily.com) for weather search functionality +- Microsoft 365 Agents Playground (optional, for testing) +- Bearer token from `a365 develop get-token -o raw` (for MCP server authentication) + +## Configuration + +### Step 1: Create Environment File + +Copy `.env.template` to `.env`: + +```bash +cp .env.template .env +``` + +### Step 2: Configure Required Variables + +**For Azure OpenAI (recommended):** +```env +AZURE_API_KEY=your-azure-openai-key +AZURE_API_BASE=https://your-resource.openai.azure.com/ +AZURE_API_VERSION=2025-01-01-preview +OPENAI_API_KEY=your-azure-openai-key # CrewAI requires this +OPENAI_MODEL_NAME=azure/gpt-4.1 # Use azure/ prefix for LiteLLM +``` + +**For OpenAI:** +```env +OPENAI_API_KEY=sk-your-openai-key +OPENAI_MODEL_NAME=gpt-4o-mini +``` + +**For MCP Tools (Mail, Calendar, Copilot):** +```env +BEARER_TOKEN= +USE_AGENTIC_AUTH=false +AGENTIC_APP_ID=crewai-agent +``` + +**For Weather Tool:** +```env +TAVILY_API_KEY=tvly-your-tavily-key +``` ## Running the Agent - ## Customizing - - Add your `OPENAI_API_KEY` into `.env`. - - Modify `src/crew_agent/config/agents.yaml` and `tasks.yaml`. - - Edit `src/crew_agent/crew.py` and `src/crew_agent/main.py` for your logic. +### Option 1: Run via Agent Runner (Standalone) + +```bash +uv run python -m crew_agent.agent_runner "London" +# or +uv run agent_runner "San Francisco, CA" +``` + +### Option 2: Hosted via Generic Agent Host (Recommended) + +This option integrates with Microsoft 365 Agents SDK for full observability and MCP tooling. - ## Running the Project +1. Ensure `.env` is configured (see above) - ### Option 1: Run via Agent Runner +2. Get a bearer token for MCP authentication: ```bash - python -m crew_agent.agent_runner "London" - # or - agent_runner "San Francisco, CA" + a365 develop get-token -o raw ``` + Copy the token to `BEARER_TOKEN` in `.env` + +3. Start the agent host: + ```bash + uv run python start_with_generic_host.py + ``` + +4. Test endpoints: + - Health check: `http://localhost:3978/api/health` + - Bot Framework: `http://localhost:3978/api/messages` + +5. Connect with Agents Playground: + ```bash + agentsplayground -e "http://localhost:3978/api/messages" -c "emulator" + ``` + +### Example Prompts + +``` +Find weather in Dublin and email the information to user@example.com under the subject "Weather Report" +``` + +``` +What's the weather in Seattle? Is it safe to drive with summer tires? +``` + +## Architecture + +### Agent Workflow + +``` +┌─────────────────────────┐ ┌──────────────────────────┐ +│ Weather Information │────▶│ Driving Safety │ +│ Specialist │ │ Specialist │ +│ (Tavily + MCP Tools) │ │ (MCP Tools only) │ +└─────────────────────────┘ └──────────────────────────┘ + │ │ + ▼ ▼ + Weather Report Safety Assessment + Email +``` + +### MCP Tools Available + +When connected with a valid bearer token, the following MCP servers are available: + +| Server | Tools | Description | +|--------|-------|-------------| +| `mcp_MailTools` | 20 tools | Send/receive emails, manage drafts, attachments | +| `mcp_CalendarTools` | 12 tools | Create/manage events, check availability | +| `mcp_M365Copilot` | 1 tool | Query Microsoft 365 Copilot | + +### Observability Scopes + +All agent operations are traced with Agent 365 observability: + +- **InvokeAgentScope**: Wraps the entire agent invocation +- **InferenceScope**: Wraps LLM calls (CrewAI crew execution) +- **ExecuteToolScope**: Wraps each MCP tool execution + +## File Structure + +``` +sample_agent/ +├── .env.template # Environment template with documentation +├── start_with_generic_host.py # Main entry point +├── host_agent_server.py # Agent 365 SDK hosting + observability setup +├── agent.py # CrewAI agent wrapper with observability +├── turn_context_utils.py # Turn context utilities +├── ToolingManifest.json # MCP server definitions +└── src/crew_agent/ + ├── config/ + │ ├── agents.yaml # Agent definitions + │ └── tasks.yaml # Task definitions + ├── crew.py # CrewAI crew orchestration + └── tools/ + └── custom_tool.py # WeatherTool (Tavily) +``` + +## Troubleshooting + +### Common Issues + +| Issue | Solution | +|-------|----------| +| `No MCP tools discovered` | Ensure `BEARER_TOKEN` is set and not expired. Run `a365 develop get-token -o raw` | +| `Weather tool returns no data` | Set `TAVILY_API_KEY` in `.env` | +| `Azure OpenAI 401 error` | Verify `AZURE_API_KEY` and `AZURE_API_BASE` are correct | +| `Duplicate emails sent` | Update `tasks.yaml` - only Driving Safety agent should send emails | +| `CrewAI tracing 401` | This is expected when CrewAI cloud tracing is not configured. It can be safely ignored; local observability still works | + +### Logs to Check - ### Option 2: Hosted via Generic Agent Host with Agent 365 (Microsoft Agent365 SDK) - Mirrors the OpenAI/AgentFramework samples and adds Agent 365 observability + MCP server registration. - - 1) Copy `.env.template` to `.env` and fill: - - `CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID` / `CLIENTSECRET` / `TENANTID` - - `AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__AGENTIC__SETTINGS__TYPE` / `SCOPES` (already in template) - - `OBSERVABILITY_SERVICE_NAME` / `OBSERVABILITY_SERVICE_NAMESPACE` - - Optional: `BEARER_TOKEN` if you want agentic auth via bearer - 2) Run host: `python start_with_generic_host.py` - 3) Health: `http://localhost:3978/api/health` (auto-fallback to next port if busy) - 4) Playground/Bot Framework endpoint: `http://localhost:3978/api/messages` - - Message text is forwarded into your CrewAI flow as the `location` input. - - MCP servers from the Agent 365 platform are discovered and passed to CrewAI agents (SSE transport with bearer headers). - - Agent 365 observability: - - Configured via `microsoft_agents_a365.observability.core.configure` in `agent.py` - - Agentic token is cached/resolved for the exporter (see `token_cache.py`) - - Startup logs show a masked env snapshot - -## Understanding Your Crew -1) **Weather Checker**: Uses web search (Tavily) to find current weather conditions -2) **Driving Safety Advisor**: Assesses whether it's safe to drive an MX-5 with summer tires based on weather data +``` +INFO:agent:✅ 33 MCP tool(s) available: # MCP connected successfully +INFO:agent:📊 Created observable wrapper # Tool wrappers with ExecuteToolScope +INFO:agent:🔧 Calling MCP tool: SendEmail... # Tool execution +``` ## Support @@ -63,7 +187,7 @@ For issues, questions, or feedback: - **CrewAI Documentation**: https://docs.crewai.com - **CrewAI GitHub**: https://github.com/joaomdmoura/crewai -- **Microsoft Agent 365 Documentation:**: https://learn.microsoft.com/en-us/microsoft-agent-365/developer/ +- **Microsoft Agent 365 Documentation**: https://learn.microsoft.com/en-us/microsoft-agent-365/developer/ - **Discord**: https://discord.com/invite/X4JWnZnxPb ## Contributing diff --git a/python/crewai/sample_agent/ToolingManifest.json b/python/crewai/sample_agent/ToolingManifest.json index 0f4ac7d6..c5c142d9 100644 --- a/python/crewai/sample_agent/ToolingManifest.json +++ b/python/crewai/sample_agent/ToolingManifest.json @@ -6,6 +6,20 @@ "url": "https://agent365.svc.cloud.microsoft/agents/servers/mcp_MailTools", "scope": "McpServers.Mail.All", "audience": "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1" + }, + { + "mcpServerName": "mcp_M365Copilot", + "mcpServerUniqueName": "mcp_M365Copilot", + "url": "https://agent365.svc.cloud.microsoft/agents/servers/mcp_M365Copilot", + "scope": "McpServers.CopilotMCP.All", + "audience": "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1" + }, + { + "mcpServerName": "mcp_CalendarTools", + "mcpServerUniqueName": "mcp_CalendarTools", + "url": "https://agent365.svc.cloud.microsoft/agents/servers/mcp_CalendarTools", + "scope": "McpServers.Calendar.All", + "audience": "ea9ffc3e-8a23-4a7d-836d-234d7c7565c1" } ] } diff --git a/python/crewai/sample_agent/agent.py b/python/crewai/sample_agent/agent.py index 20ee3a4d..dcfd98eb 100644 --- a/python/crewai/sample_agent/agent.py +++ b/python/crewai/sample_agent/agent.py @@ -1,15 +1,17 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ CrewAI Agent wrapper hosted with Microsoft Agents SDK. This keeps the CrewAI logic inside src/crew_agent and wraps it with: - Generic host contract (AgentInterface) -- Complete observability with BaggageBuilder, InvokeAgentScope, InferenceScope -- Optional MCP server discovery (metadata only today) +- Complete observability with BaggageBuilder, InvokeAgentScope, InferenceScope, ExecuteToolScope +- Full MCP server discovery, connection, and tool execution """ import asyncio +import contextvars import logging import os @@ -23,18 +25,23 @@ logger = logging.getLogger(__name__) # ============================================================================= -# DEPENDENCY IMPORTS +# IMPORTS # ============================================================================= -# # Agent Interface from agent_interface import AgentInterface # Microsoft Agents SDK from local_authentication_options import LocalAuthenticationOptions -from mcp_tool_registration_service import McpToolRegistrationService +from mcp_tool_registration_service import McpToolRegistrationService, MCPToolDefinition from microsoft_agents.hosting.core import Authorization, TurnContext +# MCP Observable Tools +from mcp_observable_tools import MCPToolExecutor, create_observable_mcp_tools + +# Notification Handler +from notification_handler import handle_notification + # Observability Components from microsoft_agents_a365.observability.core import ( InvokeAgentScope, @@ -42,10 +49,6 @@ InferenceCallDetails, InferenceOperationType, ) -from microsoft_agents_a365.observability.core.middleware.baggage_builder import BaggageBuilder - -# MCP Tooling -from microsoft_agents_a365.tooling.utils.utility import get_mcp_platform_authentication_scope # Shared turn context utilities from turn_context_utils import ( @@ -55,109 +58,94 @@ create_caller_details, create_tenant_details, create_request, + build_baggage_builder, ) +from constants import DEFAULT_AGENT_ID -# +# Context variables for thread/async-safe observability context +# These are used by MCP tool wrappers to access the current request's context +# without risk of concurrent request interference +_current_agent_details: contextvars.ContextVar = contextvars.ContextVar('agent_details', default=None) +_current_tenant_details: contextvars.ContextVar = contextvars.ContextVar('tenant_details', default=None) class CrewAIAgent(AgentInterface): """CrewAI Agent wrapper suitable for GenericAgentHost.""" - # ========================================================================= - # INITIALIZATION - # ========================================================================= - # - def __init__(self): self.auth_options = LocalAuthenticationOptions.from_environment() self.mcp_service = McpToolRegistrationService(logger=logger) + self.mcp_tool_executor = MCPToolExecutor(self.mcp_service) self.mcp_servers_initialized = False - self.mcp_servers = [] + self.mcp_tools: list[MCPToolDefinition] = [] - self._log_env_configuration() - - # Observability is already configured at module level via observability_config.py - # No need to configure again here - logger.info("CrewAI Agent uses observability configured at module level") - - # - - def _log_env_configuration(self): - """Log environment configuration for debugging.""" - logger.info("CrewAI Agent Configuration:") - logger.info(" - OPENAI_API_KEY: %s", "***" if os.getenv("OPENAI_API_KEY") else "NOT SET") - logger.info(" - AZURE_OPENAI_API_KEY: %s", "***" if os.getenv("AZURE_OPENAI_API_KEY") else "NOT SET") - logger.info(" - AZURE_OPENAI_ENDPOINT: %s", os.getenv("AZURE_OPENAI_ENDPOINT", "NOT SET")) - logger.info(" - USE_AGENTIC_AUTH: %s", os.getenv("USE_AGENTIC_AUTH", "false")) - logger.info(" - AGENTIC_APP_ID: %s", os.getenv("AGENTIC_APP_ID", "crewai-agent")) + logger.info("CrewAIAgent initialized") # ========================================================================= # MCP SERVER SETUP # ========================================================================= - # - async def _setup_mcp_servers(self, auth: Authorization, auth_handler_name: str, context: TurnContext): - """Fetch MCP server configs and convert to CrewAI MCP definitions.""" + async def _setup_mcp_servers( + self, auth: Authorization, auth_handler_name: str, context: TurnContext + ): + """ + Discover MCP servers, connect to them, and fetch available tools. + """ if self.mcp_servers_initialized: return try: - use_agentic_auth = os.getenv("USE_AGENTIC_AUTH", "false").lower() == "true" + # Get agentic_app_id from context or environment + agentic_app_id = None + if context.activity and context.activity.recipient: + agentic_app_id = context.activity.recipient.agentic_app_id + if not agentic_app_id: + agentic_app_id = os.getenv("AGENTIC_APP_ID", DEFAULT_AGENT_ID) + + # Get auth token - prefer token exchange for proper MCP authentication + use_agentic_auth = os.getenv("USE_AGENTIC_AUTH", "true").lower() == "true" auth_token = None - - if use_agentic_auth: - # Fetch token for MCP platform and pass it through - scopes = get_mcp_platform_authentication_scope() - token_obj = await auth.exchange_token( - context, - scopes=scopes, - auth_handler_id=auth_handler_name, - ) - auth_token = token_obj.token - self.mcp_servers = await self.mcp_service.list_tool_servers( - agentic_app_id=os.getenv("AGENTIC_APP_ID", "crewai-agent"), - auth=auth, - context=context, - auth_token=auth_token, - ) - else: + + if not use_agentic_auth: auth_token = self.auth_options.bearer_token - self.mcp_servers = await self.mcp_service.list_tool_servers( - agentic_app_id=os.getenv("AGENTIC_APP_ID", "crewai-agent"), - auth=auth, - context=context, - auth_token=auth_token, - ) - - mcp_entries = [] - for server in self.mcp_servers: - server_url = getattr(server, "url", None) or getattr(server, "mcp_server_unique_name", None) - server_id = getattr(server, "mcp_server_name", None) or getattr(server, "mcp_server_unique_name", None) - if not server_url or not server_id: - continue - mcp_entries.append( - { - "id": server_id, - "transport": "sse", - "options": { - "url": server_url, - "headers": {"Authorization": f"Bearer {auth_token}"} if auth_token else {}, - }, - } - ) - self.mcp_servers = mcp_entries - logger.info("MCP setup completed with %d servers (CrewAI formatted)", len(self.mcp_servers)) + logger.info("ℹ️ Using static bearer token for MCP (USE_AGENTIC_AUTH=false)") + else: + logger.info("ℹ️ MCP will use token exchange for authentication") + + # Discover and connect to MCP servers + self.mcp_tools = await self.mcp_service.discover_and_connect_servers( + agentic_app_id=agentic_app_id, + auth=auth, + auth_handler_name=auth_handler_name, + context=context, + auth_token=auth_token, + ) + + if self.mcp_tools: + logger.info(f"✅ {len(self.mcp_tools)} MCP tool(s) available:") + for tool in self.mcp_tools: + logger.info(f" 🔧 {tool.name}: {tool.description[:50]}...") + else: + logger.info("ℹ️ No MCP tools discovered") + except Exception as e: - logger.warning("MCP setup error: %s", e) + logger.error(f"Error setting up MCP servers: {e}") + self.mcp_tools = [] finally: self.mcp_servers_initialized = True - # + def _create_observable_tools(self) -> list: + """Create observable MCP tool wrappers with ExecuteToolScope tracing.""" + return create_observable_mcp_tools( + mcp_tools=self.mcp_tools, + tool_executor=self.mcp_tool_executor, + get_agent_details=lambda: _current_agent_details.get(), + get_tenant_details=lambda: _current_tenant_details.get(), + ) # ========================================================================= # MESSAGE PROCESSING # ========================================================================= - # async def initialize(self): """Initialize the agent (no-op for CrewAI wrapper).""" @@ -168,95 +156,102 @@ async def process_user_message( ) -> str: """ Process a user message by running the CrewAI flow with observability tracing. - - The message is treated as the location/prompt input to the crew. """ - # Extract context details using shared utility ctx_details = extract_turn_context_details(context) try: logger.info(f"Processing message: {message[:100]}...") - # Use BaggageBuilder to set contextual information that flows through all spans - with ( - BaggageBuilder() - .tenant_id(ctx_details.tenant_id) - .agent_id(ctx_details.agent_id) - .correlation_id(ctx_details.correlation_id) - .build() - ): - # Create observability details using shared utility + with build_baggage_builder(context, ctx_details.correlation_id).build(): + # Create observability details agent_details = create_agent_details(ctx_details, "AI agent powered by CrewAI framework") caller_details = create_caller_details(ctx_details) tenant_details = create_tenant_details(ctx_details) request = create_request(ctx_details, message) invoke_details = create_invoke_agent_details(ctx_details, "AI agent powered by CrewAI framework") - # Use context manager pattern per documentation with InvokeAgentScope.start( invoke_agent_details=invoke_details, tenant_details=tenant_details, request=request, caller_details=caller_details, ) as invoke_scope: - # Record input message if hasattr(invoke_scope, 'record_input_messages'): invoke_scope.record_input_messages([message]) # Setup MCP servers await self._setup_mcp_servers(auth, auth_handler_name, context) - # Create InferenceScope for tracking LLM call - # Note: CrewAI may use multiple LLM calls internally, this wraps the overall crew execution - inference_details = InferenceCallDetails( - operationName=InferenceOperationType.CHAT, - model=os.getenv("OPENAI_MODEL", os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")), - providerName="CrewAI (OpenAI/Azure)", - ) - - with InferenceScope.start( - details=inference_details, - agent_details=agent_details, - tenant_details=tenant_details, - request=request, - ) as inference_scope: - # Run the crew synchronously in a thread to avoid blocking the event loop - from crew_agent.agent_runner import run_crew - - logger.info("Running CrewAI with input: %s", message) - result = await asyncio.to_thread( - run_crew, - message, - True, - False, - self.mcp_servers, + # Store context for observable MCP tool wrappers using context variables + # This is thread/async-safe for concurrent request handling + agent_details_token = _current_agent_details.set(agent_details) + tenant_details_token = _current_tenant_details.set(tenant_details) + + try: + # Create observable MCP tool wrappers + observable_mcp_tools = self._create_observable_tools() + if observable_mcp_tools: + logger.info(f"📊 Created {len(observable_mcp_tools)} observable MCP tool wrapper(s)") + for tool in observable_mcp_tools: + logger.info(f" 🔧 {tool.name}: ExecuteToolScope enabled") + + # Run CrewAI with InferenceScope + full_response = await self._run_crew_with_inference_scope( + message, observable_mcp_tools, agent_details, tenant_details, request ) - logger.info("CrewAI completed") - - full_response = self._extract_result(result) - - # Record finish reasons on inference scope - if hasattr(inference_scope, 'record_finish_reasons'): - inference_scope.record_finish_reasons(["end_turn"]) - - # Record output messages on inference scope - if hasattr(inference_scope, 'record_output_messages'): - inference_scope.record_output_messages([full_response]) - # Record output message on invoke scope (after inference scope closes) - if hasattr(invoke_scope, 'record_output_messages'): - invoke_scope.record_output_messages([full_response]) + if hasattr(invoke_scope, 'record_output_messages'): + invoke_scope.record_output_messages([full_response]) + finally: + # Reset context variables to previous values + _current_agent_details.reset(agent_details_token) + _current_tenant_details.reset(tenant_details_token) - logger.info("Observability scopes closed successfully") + logger.info("✅ Observability scopes closed successfully") return full_response except Exception as e: logger.error("Error processing message: %s", e) logger.exception("Full error details:") - # Note: Context managers handle scope closure automatically, including error recording return f"Sorry, I encountered an error: {str(e)}" - # + async def _run_crew_with_inference_scope( + self, message: str, observable_mcp_tools: list, agent_details, tenant_details, request + ) -> str: + """Run CrewAI with InferenceScope for LLM call tracking.""" + inference_details = InferenceCallDetails( + operationName=InferenceOperationType.CHAT, + model=os.getenv("OPENAI_MODEL", os.getenv("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")), + providerName="CrewAI (OpenAI/Azure)", + ) + + with InferenceScope.start( + details=inference_details, + agent_details=agent_details, + tenant_details=tenant_details, + request=request, + ) as inference_scope: + from crew_agent.agent_runner import run_crew + + logger.info("Running CrewAI with input: %s", message) + result = await asyncio.to_thread( + run_crew, + message, + True, + False, + observable_mcp_tools, + ) + logger.info("CrewAI completed") + + full_response = self._extract_result(result) + + if hasattr(inference_scope, 'record_finish_reasons'): + inference_scope.record_finish_reasons(["end_turn"]) + + if hasattr(inference_scope, 'record_output_messages'): + inference_scope.record_output_messages([full_response]) + + return full_response def _extract_result(self, result) -> str: """Extract text content from crew result.""" @@ -266,18 +261,39 @@ def _extract_result(self, result) -> str: return result return str(result) + # ========================================================================= + # NOTIFICATION HANDLING + # ========================================================================= + + async def handle_agent_notification_activity( + self, + notification_activity, + auth: Authorization, + context: TurnContext, + auth_handler_name: str | None = None, + ) -> str: + """Handle agent notification activities (email, Word mentions, etc.)""" + return await handle_notification( + agent=self, + notification_activity=notification_activity, + auth=auth, + context=context, + auth_handler_name=auth_handler_name, + ) + # ========================================================================= # CLEANUP # ========================================================================= - # async def cleanup(self) -> None: - """Clean up agent resources.""" + """Clean up agent resources including MCP connections.""" try: logger.info("Cleaning up CrewAI agent resources...") - # CrewAI doesn't require explicit cleanup + + if hasattr(self, 'mcp_service'): + await self.mcp_service.cleanup() + logger.info("MCP tool registration service cleaned up") + logger.info("CrewAIAgent cleanup completed") except Exception as e: logger.error(f"Error during cleanup: {e}") - - # diff --git a/python/crewai/sample_agent/constants.py b/python/crewai/sample_agent/constants.py new file mode 100644 index 00000000..a3b2205c --- /dev/null +++ b/python/crewai/sample_agent/constants.py @@ -0,0 +1,15 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Shared Constants + +Centralized constants used across the CrewAI agent sample. +""" + +# Default agent identifier used when AGENT_ID or AGENTIC_APP_ID env vars are not set +DEFAULT_AGENT_ID = "crewai-agent" + +# Default service identifiers for observability +DEFAULT_SERVICE_NAME = "crewai-agent-sample" +DEFAULT_SERVICE_NAMESPACE = "agent365-samples" diff --git a/python/crewai/sample_agent/host_agent_server.py b/python/crewai/sample_agent/host_agent_server.py index 2b26b4d5..79e0afb9 100644 --- a/python/crewai/sample_agent/host_agent_server.py +++ b/python/crewai/sample_agent/host_agent_server.py @@ -1,8 +1,14 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ Generic Agent Host Server for CrewAI wrapper. +Features: +- Microsoft 365 Agents SDK hosting +- Observability with BaggageBuilder and InvokeAgentScope +- Notification handling (Email, Word @mentions, etc.) +- MCP tooling support """ import logging @@ -33,10 +39,19 @@ ) from microsoft_agents_a365.observability.core.middleware.baggage_builder import BaggageBuilder from microsoft_agents_a365.observability.core import InvokeAgentScope +from microsoft_agents_a365.observability.core.config import configure as configure_observability from microsoft_agents_a365.runtime.environment_utils import ( get_observability_authentication_scope, ) -from token_cache import cache_agentic_token + +# Notifications imports +from microsoft_agents_a365.notifications.agent_notification import ( + AgentNotification, + AgentNotificationActivity, + ChannelId, +) +from microsoft_agents_a365.notifications import EmailResponse, NotificationTypes + from turn_context_utils import ( extract_turn_context_details, create_invoke_agent_details, @@ -44,12 +59,20 @@ create_tenant_details, create_request, ) +from token_cache import cache_agentic_token, get_cached_agentic_token +from constants import DEFAULT_SERVICE_NAME, DEFAULT_SERVICE_NAMESPACE # Configure logging +logging.basicConfig(level=logging.INFO) + ms_agents_logger = logging.getLogger("microsoft_agents") ms_agents_logger.addHandler(logging.StreamHandler()) ms_agents_logger.setLevel(logging.INFO) +# Enable observability SDK logging to see exporter warnings +observability_logger = logging.getLogger("microsoft_agents_a365.observability") +observability_logger.setLevel(logging.INFO) + logger = logging.getLogger(__name__) # Load configuration @@ -88,6 +111,10 @@ def __init__(self, agent_class: type[AgentInterface], *agent_args, **agent_kwarg **agents_sdk_config, ) + # Initialize notification support + self.agent_notification = AgentNotification(self.agent_app) + logger.info("✅ Notification support initialized (handlers will be registered)") + # Setup message handlers self._setup_handlers() @@ -155,21 +182,24 @@ async def on_message(context: TurnContext, _: TurnState): await context.send_activity(error_msg) return - # Only perform token exchange when authentication is configured + # Only perform token registration when authentication is configured if self.auth_configured: - exaau_token = await self.agent_app.auth.exchange_token( - context, - scopes=get_observability_authentication_scope(), - auth_handler_id=self.auth_handler_name, - ) - - cache_agentic_token( - ctx_details.tenant_id, - ctx_details.agent_id, - exaau_token.token, - ) + # Exchange token and cache for sync token_resolver access + try: + exaau_token = await self.agent_app.auth.exchange_token( + context, + scopes=get_observability_authentication_scope(), + auth_handler_id=self.auth_handler_name, + ) + cache_agentic_token( + ctx_details.tenant_id, + ctx_details.agent_id, + exaau_token.token, + ) + except Exception as e: + logger.debug(f"Token exchange skipped: {e}") else: - logger.debug("Skipping token exchange in anonymous mode") + logger.debug("Skipping token registration in anonymous mode") user_message = context.activity.text or "" logger.info("Processing message: '%s'", user_message) @@ -208,6 +238,138 @@ async def on_message(context: TurnContext, _: TurnState): logger.error("Error processing message: %s", e) await context.send_activity(error_msg) + # Register notification handler + # Shared notification handler logic + async def handle_notification_common( + context: TurnContext, + state: TurnState, + notification_activity: AgentNotificationActivity, + ): + """Common notification handler for both 'agents' and 'msteams' channels""" + try: + logger.info(f"🔔 Notification received! Type: {context.activity.type}, Channel: {context.activity.channel_id if hasattr(context.activity, 'channel_id') else 'None'}") + + result = await self._validate_agent_and_setup_context(context) + if result is None: + return + tenant_id, agent_id = result + + with BaggageBuilder().tenant_id(tenant_id).agent_id(agent_id).build(): + await self._handle_notification_with_agent( + context, notification_activity + ) + + except Exception as e: + logger.error(f"❌ Notification error: {e}") + await context.send_activity( + f"Sorry, I encountered an error processing the notification: {str(e)}" + ) + + # Register a single handler for both 'agents' (production) and 'msteams' (testing) channels + # by applying the on_agent_notification decorator twice to the same function. This avoids + # duplicated handler implementations while still explicitly registering per channel. + @self.agent_notification.on_agent_notification( + channel_id=ChannelId(channel="agents", sub_channel="*"), + ) + @self.agent_notification.on_agent_notification( + channel_id=ChannelId(channel="msteams", sub_channel="*"), + ) + async def on_notification_agents_and_msteams( + context: TurnContext, + state: TurnState, + notification_activity: AgentNotificationActivity, + ): + """Handle notifications from both 'agents' (production) and 'msteams' (testing) channels""" + await handle_notification_common(context, state, notification_activity) + + logger.info("✅ Notification handler registered for 'agents' and 'msteams' channels") + + async def _handle_notification_with_agent( + self, context: TurnContext, notification_activity: AgentNotificationActivity + ): + """ + Handle notification with the agent instance. + + Args: + context: Turn context + notification_activity: The notification activity to process + """ + logger.info(f"📬 {notification_activity.notification_type}") + + # Check if agent supports notifications + if not hasattr(self.agent_instance, "handle_agent_notification_activity"): + logger.warning("⚠️ Agent doesn't support notifications") + await context.send_activity( + "This agent doesn't support notification handling yet." + ) + return + + # Process the notification with the agent + response = await self.agent_instance.handle_agent_notification_activity( + notification_activity, self.agent_app.auth, context + ) + + # For email notifications, wrap response in EmailResponse entity + if notification_activity.notification_type == NotificationTypes.EMAIL_NOTIFICATION: + response_activity = EmailResponse.create_email_response_activity(response) + await context.send_activity(response_activity) + return + + # Send the response for other notification types + await context.send_activity(response) + + async def _validate_agent_and_setup_context(self, context: TurnContext): + """ + Validate agent availability and setup observability context. + + Args: + context: Turn context from M365 SDK + + Returns: + Tuple of (tenant_id, agent_id) if successful, None if validation fails + """ + # Extract tenant and agent IDs + tenant_id = context.activity.recipient.tenant_id if context.activity.recipient else None + agent_id = context.activity.recipient.agentic_app_id if context.activity.recipient else None + + # Ensure agent is available + if not self.agent_instance: + logger.error("Agent not available") + await context.send_activity("❌ Sorry, the agent is not available.") + return None + + # Setup observability token if available + if tenant_id and agent_id: + await self._setup_observability_token(context, tenant_id, agent_id) + + return tenant_id, agent_id + + async def _setup_observability_token( + self, context: TurnContext, tenant_id: str, agent_id: str + ): + """ + Cache observability token for Agent365 exporter. + + Args: + context: Turn context + tenant_id: Tenant identifier + agent_id: Agent identifier + """ + if not self.auth_configured: + return + + try: + # Exchange token and cache for sync token_resolver access + exaau_token = await self.agent_app.auth.exchange_token( + context, + scopes=get_observability_authentication_scope(), + auth_handler_id=self.auth_handler_name, + ) + cache_agentic_token(tenant_id, agent_id, exaau_token.token) + logger.debug(f"✅ Cached observability token for {tenant_id}:{agent_id}") + except Exception as e: + logger.warning(f"⚠️ Failed to cache observability token: {e}") + async def initialize_agent(self): """Initialize the hosted agent instance.""" if self.agent_instance is None: @@ -358,9 +520,58 @@ def create_and_run_host(agent_class: type[AgentInterface], *agent_args, **agent_ if not check_agent_inheritance(agent_class): raise TypeError(f"Agent class {agent_class.__name__} must inherit from AgentInterface") - # Initialize observability before starting the host - from observability_config import initialize_observability - initialize_observability() + # Configure observability if not already configured (e.g., by start_with_generic_host.py) + # Note: Early configuration in start_with_generic_host.py is preferred to avoid + # CrewAI's TracerProvider being set up before ours + enable_observability = os.getenv("ENABLE_OBSERVABILITY", "true").lower() in ("true", "1", "yes") + if enable_observability: + from opentelemetry import trace as otel_trace + existing_provider = otel_trace.get_tracer_provider() + provider_type = type(existing_provider).__name__ + + # Check if observability was already configured with our service name + is_already_configured = False + if hasattr(existing_provider, 'resource'): + resource_attrs = dict(existing_provider.resource.attributes) + service_name = resource_attrs.get('service.name', '') + # If service name contains our identifier, skip reconfiguration + if DEFAULT_SERVICE_NAME.split('-')[0] in service_name.lower() or 'agent365' in service_name.lower(): + is_already_configured = True + logger.info(f"✅ Observability already configured: {service_name}") + + if not is_already_configured: + service_name = os.getenv("OBSERVABILITY_SERVICE_NAME", DEFAULT_SERVICE_NAME) + service_namespace = os.getenv("OBSERVABILITY_SERVICE_NAMESPACE", DEFAULT_SERVICE_NAMESPACE) + + # Token resolver for observability exporter (must be sync) + def token_resolver(agent_id: str, tenant_id: str) -> str | None: + """Resolve authentication token for observability exporter""" + token = get_cached_agentic_token(tenant_id, agent_id) + if token: + logger.debug(f"Token resolver: found cached token for {agent_id}:{tenant_id}") + else: + logger.debug(f"Token resolver: no cached token for {agent_id}:{tenant_id}") + return token + + try: + logger.info(f"🔍 Existing TracerProvider: {provider_type}") + if hasattr(existing_provider, 'resource'): + logger.info(f"🔍 Existing resource: {existing_provider.resource.attributes}") + + configure_observability( + service_name=service_name, + service_namespace=service_namespace, + token_resolver=token_resolver, + cluster_category=os.getenv("PYTHON_ENVIRONMENT", "development"), + ) + print("✅ Observability configured") + logger.info(f"✅ Observability configured: {service_name} ({service_namespace})") + except Exception as e: + print(f"⚠️ Failed to configure observability: {e}") + logger.warning(f"⚠️ Failed to configure observability: {e}") + else: + print("ℹ️ Observability disabled (ENABLE_OBSERVABILITY=false)") + logger.info("ℹ️ Observability disabled (ENABLE_OBSERVABILITY=false)") host = GenericAgentHost(agent_class, *agent_args, **agent_kwargs) auth_config = host.create_auth_configuration() diff --git a/python/crewai/sample_agent/mcp_observable_tools.py b/python/crewai/sample_agent/mcp_observable_tools.py new file mode 100644 index 00000000..ee261ac7 --- /dev/null +++ b/python/crewai/sample_agent/mcp_observable_tools.py @@ -0,0 +1,264 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +MCP Observable Tools Module + +Creates CrewAI-compatible tool wrappers for MCP tools with ExecuteToolScope observability. +This module separates the MCP tool wrapper logic from the main agent for better readability. +""" + +import asyncio +import json +import logging +import uuid +from typing import TYPE_CHECKING, Any, Type +from urllib.parse import urlparse + +from crewai.tools import BaseTool +from pydantic import BaseModel, Field, create_model + +from microsoft_agents_a365.observability.core import ExecuteToolScope, ToolCallDetails +from mcp_tool_registration_service import MCPToolDefinition + +if TYPE_CHECKING: + from mcp_tool_registration_service import McpToolRegistrationService + +logger = logging.getLogger(__name__) + + +class MCPToolExecutor: + """ + Handles MCP tool execution with observability tracing. + + This class encapsulates the logic for calling MCP tools with proper + ExecuteToolScope observability, separated from the main agent class. + """ + + def __init__(self, mcp_service: "McpToolRegistrationService"): + """ + Initialize the MCP tool executor. + + Args: + mcp_service: The MCP tool registration service for executing tools + """ + self.mcp_service = mcp_service + + async def call_tool( + self, + tool_name: str, + arguments: dict, + agent_details: Any, + tenant_details: Any, + ) -> str: + """ + Call an MCP tool by name with ExecuteToolScope observability. + + Args: + tool_name: Name of the tool to call + arguments: Tool arguments as a dictionary + agent_details: AgentDetails for observability + tenant_details: TenantDetails for observability + + Returns: + The tool result as a string + """ + tool = self.mcp_service.get_tool_by_name(tool_name) + tool_call_id = str(uuid.uuid4()) + + # Serialize arguments for observability + try: + args_str = json.dumps(arguments) if arguments else "" + except (TypeError, ValueError): + args_str = str(arguments) if arguments else "" + + # Determine endpoint URL + endpoint_url = tool.server_url if tool else "" + endpoint = urlparse(endpoint_url) if endpoint_url else None + + # Create ToolCallDetails for observability + tool_call_details = ToolCallDetails( + tool_name=tool_name, + arguments=args_str, + tool_call_id=tool_call_id, + description=f"Executing MCP tool: {tool_name}", + tool_type="mcp_extension", + endpoint=endpoint, + ) + + # Execute with ExecuteToolScope for observability + with ExecuteToolScope.start( + details=tool_call_details, + agent_details=agent_details, + tenant_details=tenant_details, + ) as tool_scope: + try: + logger.info(f"🔧 Calling MCP tool: {tool_name}") + result = await self.mcp_service.call_tool(tool_name, arguments) + + # Record the response + if hasattr(tool_scope, 'record_response'): + tool_scope.record_response(str(result) if result else "") + + logger.info(f"✅ MCP tool '{tool_name}' executed successfully") + return result + + except Exception as e: + logger.error(f"❌ MCP tool '{tool_name}' failed: {e}") + raise + + +def create_observable_mcp_tools( + mcp_tools: list[MCPToolDefinition], + tool_executor: MCPToolExecutor, + get_agent_details: callable, + get_tenant_details: callable, +) -> list[BaseTool]: + """ + Create CrewAI-compatible tool wrappers for MCP tools with ExecuteToolScope observability. + + Instead of using CrewAI's native `mcps` parameter (which handles tool execution internally + without observability hooks), this function creates custom CrewAI tools that wrap MCP calls + with ExecuteToolScope for proper tracing. + + Args: + mcp_tools: List of MCP tool definitions from the registration service + tool_executor: MCPToolExecutor instance for executing tools with observability + get_agent_details: Callable that returns current agent details for observability + get_tenant_details: Callable that returns current tenant details for observability + + Returns: + List of CrewAI BaseTool instances that wrap MCP tools with observability + """ + observable_tools = [] + + for mcp_tool in mcp_tools: + # Create dynamic input schema from MCP tool's input schema + input_fields = _build_input_fields(mcp_tool) + + # Create dynamic Pydantic model for input schema + InputModel = create_model(f"{mcp_tool.name}Input", **input_fields) + + # Create a closure to capture the tool definition and executor reference + tool_class = _create_tool_class( + mcp_tool, InputModel, tool_executor, get_agent_details, get_tenant_details + ) + + observable_tools.append(tool_class) + logger.info(f"📊 Created observable wrapper for MCP tool: {mcp_tool.name}") + + return observable_tools + + +def _build_input_fields(mcp_tool: MCPToolDefinition) -> dict: + """ + Build Pydantic field definitions from MCP tool input schema. + + Args: + mcp_tool: The MCP tool definition + + Returns: + Dictionary of field name to (type, Field) tuples for Pydantic model creation + """ + input_fields = {} + + if mcp_tool.input_schema and "properties" in mcp_tool.input_schema: + for prop_name, prop_def in mcp_tool.input_schema.get("properties", {}).items(): + field_type = _get_python_type(prop_def.get("type")) + description = prop_def.get("description", f"Parameter {prop_name}") + required = prop_name in mcp_tool.input_schema.get("required", []) + + if required: + input_fields[prop_name] = (field_type, Field(..., description=description)) + else: + input_fields[prop_name] = (field_type, Field(default=None, description=description)) + + # If no input schema, create a generic one + if not input_fields: + input_fields["input"] = (str, Field(default="", description="Input for the tool")) + + return input_fields + + +def _get_python_type(json_type: str | None) -> type: + """ + Convert JSON schema type to Python type. + + Args: + json_type: JSON schema type string + + Returns: + Corresponding Python type + """ + type_mapping = { + "integer": int, + "number": float, + "boolean": bool, + "array": list, + "object": dict, + } + return type_mapping.get(json_type, str) + + +def _create_tool_class( + tool_def: MCPToolDefinition, + input_model: Type[BaseModel], + tool_executor: MCPToolExecutor, + get_agent_details: callable, + get_tenant_details: callable, +) -> BaseTool: + """ + Create a CrewAI BaseTool class for an MCP tool with observability. + + Args: + tool_def: The MCP tool definition + input_model: Pydantic model for input validation + tool_executor: MCPToolExecutor for executing the tool + get_agent_details: Callable to get current agent details + get_tenant_details: Callable to get current tenant details + + Returns: + Instance of the created tool class + """ + class ObservableMCPTool(BaseTool): + name: str = tool_def.name + description: str = tool_def.description or f"MCP tool: {tool_def.name}" + args_schema: Type[BaseModel] = input_model + + def _run(self, **kwargs) -> str: + """Execute MCP tool with ExecuteToolScope observability.""" + # Run async call_tool in sync context + # Handle both cases: running inside an existing event loop or not + try: + # Check if there's already a running event loop (e.g., from CrewAI) + loop = asyncio.get_running_loop() + # We're inside an async context - use run_coroutine_threadsafe + # This shouldn't normally happen with CrewAI's sync tool execution + import concurrent.futures + future = asyncio.run_coroutine_threadsafe( + tool_executor.call_tool( + tool_name=tool_def.name, + arguments=kwargs, + agent_details=get_agent_details(), + tenant_details=get_tenant_details(), + ), + loop + ) + result = future.result(timeout=300) # 5 minute timeout + return result + except RuntimeError: + # No running event loop - use asyncio.run() for clean lifecycle management + result = asyncio.run( + tool_executor.call_tool( + tool_name=tool_def.name, + arguments=kwargs, + agent_details=get_agent_details(), + tenant_details=get_tenant_details(), + ) + ) + return result + except Exception as e: + logger.error(f"❌ MCP tool '{tool_def.name}' error: {e}") + return f"Error executing {tool_def.name}: {str(e)}" + + return ObservableMCPTool() diff --git a/python/crewai/sample_agent/mcp_tool_registration_service.py b/python/crewai/sample_agent/mcp_tool_registration_service.py index fd264a02..ebdbefe5 100644 --- a/python/crewai/sample_agent/mcp_tool_registration_service.py +++ b/python/crewai/sample_agent/mcp_tool_registration_service.py @@ -1,15 +1,32 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ -Lightweight MCP tool registration for CrewAI wrapper. +MCP Tool Registration Service for CrewAI Agent + +This service provides MCP (Model Context Protocol) tool integration for CrewAI agents, +enabling tool discovery, connection, and execution against Agent365 MCP servers. +Features: +- Discovers MCP servers from McpToolServerConfigurationService (production) or ToolingManifest.json (dev) +- Connects to MCP servers via Streamable HTTP +- Lists available tools from MCP servers +- Executes MCP tool calls and returns results +- Handles authentication and authorization +- Converts tools to CrewAI-compatible format """ -from typing import Optional, List +from typing import Dict, List, Optional, Any +from dataclasses import dataclass, field import logging +import os +import random +import aiohttp +import asyncio +import json from microsoft_agents.hosting.core import Authorization, TurnContext - +from microsoft_agents_a365.tooling.utils.constants import Constants from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import ( McpToolServerConfigurationService, ) @@ -17,14 +34,623 @@ get_mcp_platform_authentication_scope, ) +# Default MCP Platform endpoint (can be overridden via MCP_PLATFORM_ENDPOINT env var) +DEFAULT_MCP_PLATFORM_ENDPOINT = "https://agent365.svc.cloud.microsoft" + +# Production configuration +MCP_REQUEST_TIMEOUT_SECONDS = 30 +MCP_CONNECT_TIMEOUT_SECONDS = 10 +MCP_MAX_RETRIES = 2 +MCP_RETRY_BASE_DELAY_SECONDS = 1 # Base delay for exponential backoff + + +def get_mcp_platform_endpoint() -> str: + """Get the MCP platform endpoint from environment or use default.""" + endpoint = os.getenv("MCP_PLATFORM_ENDPOINT", "").strip() + return endpoint if endpoint else DEFAULT_MCP_PLATFORM_ENDPOINT + + +@dataclass +class MCPToolDefinition: + """Definition of an MCP tool""" + name: str + description: str + input_schema: Dict[str, Any] + server_url: str + server_name: str + + +@dataclass +class MCPServerConnection: + """Information about a connected MCP server""" + name: str + url: str + headers: Dict[str, str] = field(default_factory=dict) + tools: List[MCPToolDefinition] = field(default_factory=list) + connected: bool = False + class McpToolRegistrationService: - """Service for listing MCP tool servers for an agentic app id.""" + """ + Service for managing MCP tools and servers for CrewAI agents. + + This service provides tool discovery, connection, and execution capabilities + for MCP servers, enabling CrewAI agents to use Agent365 platform tools. + + Discovery modes: + - Production: Uses McpToolServerConfigurationService to discover servers from Gateway + - Development: Falls back to ToolingManifest.json if SDK returns no servers + """ + + _orchestrator_name: str = "CrewAI" def __init__(self, logger: Optional[logging.Logger] = None): + """ + Initialize the MCP Tool Registration Service for CrewAI. + + Args: + logger: Logger instance for logging operations. + """ self._logger = logger or logging.getLogger(self.__class__.__name__) - self.config_service = McpToolServerConfigurationService(logger=self._logger) + self._connected_servers: List[MCPServerConnection] = [] + self._tools_by_name: Dict[str, MCPToolDefinition] = {} + self._auth_token: Optional[str] = None + self._config_service = McpToolServerConfigurationService(logger=self._logger) + + def _load_manifest_servers_fallback(self) -> List[Dict[str, Any]]: + """ + Load MCP server configurations directly from ToolingManifest.json. + + This is a fallback for local development when McpToolServerConfigurationService + cannot discover servers (e.g., no Gateway connection). + + Returns: + List of server configurations with name, url, scope, audience. + """ + servers = [] + manifest_path = os.path.join(os.getcwd(), "ToolingManifest.json") + + if not os.path.exists(manifest_path): + self._logger.debug(f"ToolingManifest.json not found at {manifest_path}") + return servers + + try: + with open(manifest_path, 'r') as f: + manifest = json.load(f) + + self._logger.info(f"📄 [Fallback] Loaded ToolingManifest.json") + + for server in manifest.get("mcpServers", []): + name = server.get("mcpServerName", server.get("mcpServerUniqueName", "unknown")) + url = server.get("url", "") + scope = server.get("scope", "") + audience = server.get("audience", "") + + if url: + servers.append({ + "name": name, + "url": url, + "scope": scope, + "audience": audience, + }) + self._logger.info(f" 📌 [Manifest] Server: {name} -> {url}") + + except Exception as e: + self._logger.error(f"Failed to load ToolingManifest.json: {e}") + + return servers + + def _build_full_url(self, server_path: str) -> str: + """ + Build a full URL from a server path. + + If the path is already a full URL (http/https), return as-is. + Otherwise, prepend the MCP platform endpoint. + + Args: + server_path: The server URL or relative path + + Returns: + Full URL to the MCP server + """ + if not server_path: + return "" + + # Already a full URL + if server_path.startswith("http://") or server_path.startswith("https://"): + return server_path + + # Build full URL from relative path + platform_endpoint = get_mcp_platform_endpoint() + + # Handle different path formats: + # - "/agents/servers/mcp_MailTools" -> prepend endpoint + # - "agents/servers/mcp_MailTools" -> prepend endpoint with / + # - "mcp_MailTools" -> assume it's under /agents/servers/ + path = server_path.lstrip("/") + + if not path.startswith("agents/"): + # Just a server name like "mcp_MailTools" + path = f"agents/servers/{path}" + + return f"{platform_endpoint.rstrip('/')}/{path}" + + async def discover_and_connect_servers( + self, + agentic_app_id: str, + auth: Authorization, + auth_handler_name: str, + context: TurnContext, + auth_token: Optional[str] = None, + ) -> List[MCPToolDefinition]: + """ + Discover MCP servers using McpToolServerConfigurationService and connect to them. + + In production, uses the SDK service to discover servers from Gateway. + In development, falls back to ToolingManifest.json if SDK returns no servers. + + Args: + agentic_app_id: The agent's application ID for server discovery. + auth: Authorization handler for token exchange. + auth_handler_name: Name of the authorization handler. + context: Turn context for the current operation. + auth_token: Optional pre-configured authentication token. + + Returns: + List of all available tool definitions from connected servers. + """ + # Get authentication token if not provided + if not auth_token: + try: + scopes = get_mcp_platform_authentication_scope() + self._logger.info(f"🔑 Attempting token exchange with scopes: {scopes}") + auth_result = await auth.exchange_token(context, scopes, auth_handler_name) + if auth_result and auth_result.token: + auth_token = auth_result.token + self._logger.info("✅ Token exchange successful for MCP authentication") + else: + self._logger.warning("⚠️ Token exchange returned no token") + except Exception as e: + self._logger.warning(f"⚠️ Token exchange failed: {type(e).__name__}: {e}") + + # Fallback to static BEARER_TOKEN from environment + if not auth_token: + bearer_token = os.getenv("BEARER_TOKEN", "").strip() + if bearer_token: + auth_token = bearer_token + self._logger.info("ℹ️ Using BEARER_TOKEN from environment for MCP authentication") + + # For local development, allow connections without auth token + if not auth_token: + self._logger.info("ℹ️ No auth token - will attempt local connections only") + + self._auth_token = auth_token + + # Get the MCP platform base URL for reference + platform_endpoint = get_mcp_platform_endpoint() + self._logger.info(f"🌐 MCP Platform endpoint: {platform_endpoint}") + + # Try to discover servers using McpToolServerConfigurationService (production path) + mcp_server_configs = [] + try: + self._logger.info(f"🔍 Discovering MCP servers for agent {agentic_app_id}") + sdk_configs = await self._config_service.list_tool_servers( + agentic_app_id=agentic_app_id, + auth_token=auth_token if auth_token else None, + ) + + # Convert SDK config objects to our format + for config in sdk_configs: + # Extract URL - try different attribute names the SDK might use + server_url = getattr(config, "url", None) or \ + getattr(config, "server_url", None) or \ + getattr(config, "endpoint", None) + + server_name = getattr(config, "mcp_server_name", None) or \ + getattr(config, "mcp_server_unique_name", None) or \ + getattr(config, "name", "unknown") + + # If URL is not a full URL, it might just be the server name/path + if not server_url: + # Use server name as path if no URL provided + server_url = getattr(config, "mcp_server_unique_name", None) or server_name + + # Build full URL + full_url = self._build_full_url(server_url) + + if full_url: + mcp_server_configs.append({ + "name": server_name, + "url": full_url, + }) + self._logger.info(f" 📌 [SDK] Server: {server_name} -> {full_url}") + + self._logger.info(f"📋 SDK discovered {len(mcp_server_configs)} MCP server(s)") + + except Exception as e: + self._logger.warning(f"⚠️ McpToolServerConfigurationService failed: {e}") + + # Fallback to ToolingManifest.json if SDK returned no servers (development mode) + if not mcp_server_configs: + self._logger.info("📄 Falling back to ToolingManifest.json for server discovery") + mcp_server_configs = self._load_manifest_servers_fallback() + + self._logger.info(f"Found {len(mcp_server_configs)} MCP server configurations total") + + # Connect to each server and fetch tools + all_tools: List[MCPToolDefinition] = [] + + for server_config in mcp_server_configs: + try: + connection = await self._connect_to_server( + name=server_config["name"], + url=server_config["url"], + auth_token=auth_token, + ) + + if connection and connection.connected: + self._connected_servers.append(connection) + all_tools.extend(connection.tools) + + # Index tools by name for quick lookup + for tool in connection.tools: + self._tools_by_name[tool.name] = tool + + self._logger.info( + f"Connected to MCP server '{connection.name}' with " + f"{len(connection.tools)} tools" + ) + + except (TimeoutError, ConnectionError, OSError) as e: + # Recoverable network errors - continue to next server + self._logger.warning( + f"Failed to connect to MCP server {server_config['name']}: {e}" + ) + continue + except Exception as e: + # Non-recoverable or unexpected errors - log full stack trace + import traceback + self._logger.error( + f"Unexpected error connecting to MCP server {server_config['name']}: {e}\n" + f"{traceback.format_exc()}" + ) + continue + + self._logger.info(f"Total {len(all_tools)} MCP tools available") + return all_tools + + async def _connect_to_server( + self, + name: str, + url: str, + auth_token: str, + ) -> Optional[MCPServerConnection]: + """ + Connect to an MCP server and fetch its tools. + + Args: + name: Server display name. + url: Server URL endpoint. + auth_token: Authentication token. + + Returns: + MCPServerConnection with tools, or None if connection failed. + """ + # Check if this is a local server (no auth needed) + is_local = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") + + if is_local: + headers = { + "Content-Type": "application/json", + } + self._logger.info(f"🏠 Connecting to local MCP server: {url}") + else: + if not auth_token: + self._logger.warning(f"⚠️ Skipping remote server {name} - no auth token") + return None + headers = { + Constants.Headers.AUTHORIZATION: f"{Constants.Headers.BEARER_PREFIX} {auth_token}", + "User-Agent": f"CrewAI-Agent-SDK/1.0 ({self._orchestrator_name})", + "Content-Type": "application/json", + } + self._logger.info(f"☁️ Connecting to remote MCP server: {url}") + + connection = MCPServerConnection( + name=name, + url=url, + headers=headers, + ) + + try: + # Fetch available tools from the server + tools = await self._list_server_tools(url, headers, name) + connection.tools = tools + connection.connected = True + return connection + + except Exception as e: + self._logger.error(f"Failed to connect to MCP server {name} at {url}: {e}") + return None + + async def _parse_sse_response(self, response) -> Dict[str, Any]: + """ + Parse Server-Sent Events (SSE) response from MCP server. + + Agent 365 MCP servers use Streamable HTTP transport which returns + responses as SSE with content-type: text/event-stream. + + Args: + response: aiohttp response object + + Returns: + Parsed JSON-RPC result from the SSE stream + """ + content_type = response.headers.get('Content-Type', '') + + # If it's regular JSON, parse directly + if 'application/json' in content_type: + return await response.json() + + # Handle SSE (text/event-stream) + result = None + async for line in response.content: + line_str = line.decode('utf-8').strip() + + # Skip empty lines and comments + if not line_str or line_str.startswith(':'): + continue + + # Parse SSE data lines + if line_str.startswith('data:'): + data_str = line_str[5:].strip() + if data_str: + try: + parsed = json.loads(data_str) + # Look for the JSON-RPC result + if 'result' in parsed or 'error' in parsed: + result = parsed + break + # Some servers send the result directly + if 'jsonrpc' in parsed: + result = parsed + break + except json.JSONDecodeError: + continue + # Handle non-prefixed JSON lines (some SSE implementations) + elif line_str.startswith('{'): + try: + parsed = json.loads(line_str) + if 'result' in parsed or 'jsonrpc' in parsed: + result = parsed + break + except json.JSONDecodeError: + continue + + if result is None: + raise Exception("No valid JSON-RPC response found in SSE stream") + + return result + + async def _list_server_tools( + self, + server_url: str, + headers: Dict[str, str], + server_name: str, + ) -> List[MCPToolDefinition]: + """ + List available tools from an MCP server. + + Args: + server_url: The MCP server URL endpoint. + headers: HTTP headers including authorization. + server_name: Server name for tool attribution. + + Returns: + List of tool definitions. + """ + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/list", + "params": {} + } + + # Add Accept header for SSE + request_headers = {**headers, "Accept": "text/event-stream, application/json"} + + # Configure timeout for production + timeout = aiohttp.ClientTimeout( + total=MCP_REQUEST_TIMEOUT_SECONDS, + connect=MCP_CONNECT_TIMEOUT_SECONDS + ) + + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post(server_url, headers=request_headers, json=payload) as response: + if response.status == 200: + result = await self._parse_sse_response(response) + tools_data = result.get("result", {}).get("tools", []) + + tools = [] + for tool_data in tools_data: + tool = MCPToolDefinition( + name=tool_data.get("name", ""), + description=tool_data.get("description", ""), + input_schema=tool_data.get("inputSchema", {}), + server_url=server_url, + server_name=server_name, + ) + tools.append(tool) + + self._logger.debug(f"Listed {len(tools)} tools from {server_name}") + return tools + else: + error_text = await response.text() + raise Exception(f"Failed to list tools: {response.status} - {error_text}") + + async def call_tool( + self, + tool_name: str, + arguments: Dict[str, Any], + ) -> str: + """ + Execute an MCP tool and return the result. + + Includes retry logic for transient failures and proper timeout handling. + + Args: + tool_name: Name of the tool to execute. + arguments: Tool arguments as a dictionary. + + Returns: + The tool result as a string. + + Raises: + ValueError: If the tool is not found or not connected. + Exception: If the tool call fails after retries. + """ + if tool_name not in self._tools_by_name: + available = list(self._tools_by_name.keys())[:10] # Limit for logging + raise ValueError(f"Tool '{tool_name}' not found. Available tools: {available}...") + + tool = self._tools_by_name[tool_name] + + # Find the connection for this tool + connection = None + for conn in self._connected_servers: + if conn.url == tool.server_url: + connection = conn + break + + if not connection: + raise ValueError(f"No connection found for tool '{tool_name}'") + + payload = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": arguments + } + } + + self._logger.info(f"Calling MCP tool '{tool_name}' on server '{connection.name}'") + self._logger.debug(f"Tool arguments: {arguments}") + + # Add Accept header for SSE + request_headers = {**connection.headers, "Accept": "text/event-stream, application/json"} + + # Configure timeout for production + timeout = aiohttp.ClientTimeout( + total=MCP_REQUEST_TIMEOUT_SECONDS, + connect=MCP_CONNECT_TIMEOUT_SECONDS + ) + + last_error = None + for attempt in range(MCP_MAX_RETRIES + 1): + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post( + connection.url, + headers=request_headers, + json=payload + ) as response: + if response.status == 200: + result = await self._parse_sse_response(response) + + # Extract content from MCP response + content = result.get("result", {}).get("content", []) + if content and len(content) > 0: + # Handle different content types + first_content = content[0] + if isinstance(first_content, dict): + result_text = first_content.get("text", str(first_content)) + else: + result_text = str(first_content) + + self._logger.info(f"MCP tool '{tool_name}' executed successfully") + return result_text + + return str(result.get("result", "")) + + elif response.status in (502, 503, 504): + # Retryable server errors + error_text = await response.text() + last_error = Exception(f"MCP server error: {response.status} - {error_text}") + self._logger.warning(f"Retryable error on attempt {attempt + 1}: {response.status}") + else: + # Non-retryable error + error_text = await response.text() + raise Exception(f"MCP tool call failed: {response.status} - {error_text}") + + except asyncio.TimeoutError: + last_error = Exception(f"MCP tool call timed out after {MCP_REQUEST_TIMEOUT_SECONDS}s") + self._logger.warning(f"Timeout on attempt {attempt + 1} for tool '{tool_name}'") + except aiohttp.ClientError as e: + last_error = e + self._logger.warning(f"Connection error on attempt {attempt + 1}: {e}") + + # Wait before retry with exponential backoff and jitter (except on last attempt) + if attempt < MCP_MAX_RETRIES: + # Exponential backoff: base_delay * 2^attempt + random jitter (0-0.5s) + delay = MCP_RETRY_BASE_DELAY_SECONDS * (2 ** attempt) + random.uniform(0, 0.5) + self._logger.info(f"Retrying in {delay:.2f}s...") + await asyncio.sleep(delay) + + # All retries exhausted + self._logger.error(f"MCP tool '{tool_name}' failed after {MCP_MAX_RETRIES + 1} attempts") + raise last_error or Exception("MCP tool call failed") + + def get_tools_for_crewai(self) -> List[Dict[str, Any]]: + """ + Get tool definitions in CrewAI's expected format. + + Returns: + List of MCP server configurations formatted for CrewAI. + """ + crewai_mcp_servers = [] + + for connection in self._connected_servers: + crewai_mcp_servers.append({ + "id": connection.name, + "transport": "sse", + "options": { + "url": connection.url, + "headers": connection.headers, + }, + }) + + return crewai_mcp_servers + + def get_available_tool_names(self) -> List[str]: + """ + Get list of available MCP tool names. + + Returns: + List of tool names that can be called. + """ + return list(self._tools_by_name.keys()) + + def get_tool_by_name(self, name: str) -> Optional[MCPToolDefinition]: + """ + Get a tool definition by name. + + Args: + name: The tool name to look up. + + Returns: + MCPToolDefinition or None if not found. + """ + return self._tools_by_name.get(name) + + async def cleanup(self): + """Clean up all connected MCP servers.""" + self._connected_servers = [] + self._tools_by_name = {} + self._auth_token = None + self._logger.info("MCP tool registration service cleaned up") + # Legacy method for backwards compatibility async def list_tool_servers( self, agentic_app_id: str, @@ -35,6 +661,9 @@ async def list_tool_servers( ) -> List: """ Fetch MCP server configurations the agent is allowed to use. + + This is a legacy method for backwards compatibility. + Prefer using discover_and_connect_servers() for full functionality. Returns a list of MCP server configuration objects (metadata only). """ @@ -45,7 +674,7 @@ async def list_tool_servers( token = auth_token_obj.token self._logger.info("Listing MCP tool servers for agent %s", agentic_app_id) - mcp_server_configs = await self.config_service.list_tool_servers( + mcp_server_configs = await self._config_service.list_tool_servers( agentic_app_id=agentic_app_id, auth_token=token, ) diff --git a/python/crewai/sample_agent/notification_handler.py b/python/crewai/sample_agent/notification_handler.py new file mode 100644 index 00000000..c550620d --- /dev/null +++ b/python/crewai/sample_agent/notification_handler.py @@ -0,0 +1,151 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Notification Handler Module + +Handles agent notification activities (email, Word mentions, etc.) for the CrewAI agent. +This module separates notification handling logic from the main agent for better readability. +""" + +import logging +from typing import TYPE_CHECKING, Any + +from microsoft_agents.hosting.core import Authorization, TurnContext +from microsoft_agents_a365.notifications import NotificationTypes + +if TYPE_CHECKING: + from agent import CrewAIAgent + +logger = logging.getLogger(__name__) + + +async def handle_notification( + agent: "CrewAIAgent", + notification_activity: Any, + auth: Authorization, + context: TurnContext, + auth_handler_name: str | None = None, +) -> str: + """ + Handle agent notification activities (email, Word mentions, etc.) + + Args: + agent: The CrewAI agent instance to process messages + notification_activity: The notification activity from Agent365 + auth: Authorization for token exchange + context: Turn context from M365 SDK + auth_handler_name: Optional auth handler name for token exchange + + Returns: + Response string to send back + """ + try: + notification_type = notification_activity.notification_type + logger.info(f"📬 Processing notification: {notification_type}") + + # Validate notification_type before comparisons + if notification_type is None: + logger.warning("Received notification with None notification_type") + return await _handle_generic_notification( + agent, notification_activity, auth, context, auth_handler_name + ) + + # Handle Email Notifications + if notification_type == NotificationTypes.EMAIL_NOTIFICATION: + return await _handle_email_notification( + agent, notification_activity, auth, context, auth_handler_name + ) + + # Handle Word Comment Notifications + elif notification_type == NotificationTypes.WPX_COMMENT: + return await _handle_word_notification( + agent, notification_activity, auth, context, auth_handler_name + ) + + # Generic notification handling + else: + return await _handle_generic_notification( + agent, notification_activity, auth, context, auth_handler_name + ) + + except Exception as e: + logger.error(f"Error processing notification: {e}") + logger.exception("Full error details:") + return f"Sorry, I encountered an error processing the notification: {str(e)}" + + +async def _handle_email_notification( + agent: "CrewAIAgent", + notification_activity: Any, + auth: Authorization, + context: TurnContext, + auth_handler_name: str | None, +) -> str: + """Handle email notification activities.""" + if not hasattr(notification_activity, "email") or not notification_activity.email: + return "I could not find the email notification details." + + email = notification_activity.email + email_body = getattr(email, "html_body", "") or getattr(email, "body", "") + + message = f"You have received the following email. Please follow any instructions in it.\n\n{email_body}" + logger.info("📧 Processing email notification") + + response = await agent.process_user_message(message, auth, auth_handler_name, context) + return response or "Email notification processed." + + +async def _handle_word_notification( + agent: "CrewAIAgent", + notification_activity: Any, + auth: Authorization, + context: TurnContext, + auth_handler_name: str | None, +) -> str: + """Handle Word document comment notification activities.""" + if not hasattr(notification_activity, "wpx_comment") or not notification_activity.wpx_comment: + return "I could not find the Word notification details." + + wpx = notification_activity.wpx_comment + doc_id = getattr(wpx, "document_id", "") + comment_text = notification_activity.text or "" + + logger.info(f"📄 Processing Word comment notification for doc {doc_id}") + + message = ( + f"You have been mentioned in a Word document comment.\n" + f"Document ID: {doc_id}\n" + f"Comment: {comment_text}\n\n" + f"Please respond to this comment appropriately." + ) + + response = await agent.process_user_message(message, auth, auth_handler_name, context) + return response or "Word notification processed." + + +async def _handle_generic_notification( + agent: "CrewAIAgent", + notification_activity: Any, + auth: Authorization, + context: TurnContext, + auth_handler_name: str | None, +) -> str: + """Handle generic notification activities.""" + logger.info("🔍 Full notification activity structure:") + logger.info(f" Type: {notification_activity.activity.type}") + logger.info(f" Name: {notification_activity.activity.name}") + logger.info(f" Text: {getattr(notification_activity.activity, 'text', 'N/A')}") + logger.info(f" Value: {getattr(notification_activity.activity, 'value', 'N/A')}") + logger.info(f" Entities: {notification_activity.activity.entities}") + logger.info(f" Channel ID: {notification_activity.activity.channel_id}") + + notification_message = ( + getattr(notification_activity.activity, 'text', None) or + str(getattr(notification_activity.activity, 'value', None)) or + f"Notification received: {notification_activity.notification_type}" + ) + logger.info(f"📨 Processing generic notification: {notification_activity.notification_type}") + + response = await agent.process_user_message(notification_message, auth, auth_handler_name, context) + return response or "Notification processed successfully." diff --git a/python/crewai/sample_agent/observability_config.py b/python/crewai/sample_agent/observability_config.py deleted file mode 100644 index 6c3125c1..00000000 --- a/python/crewai/sample_agent/observability_config.py +++ /dev/null @@ -1,65 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - -""" -Observability Configuration Module - -Provides observability initialization for Agent 365 SDK. -""" - -import logging -import os - -from microsoft_agents_a365.observability.core.config import configure -from token_cache import get_cached_agentic_token - -logger = logging.getLogger(__name__) - - -def token_resolver(agent_id: str, tenant_id: str) -> str | None: - """Token resolver for Agent 365 Observability exporter.""" - try: - logger.info(f"Token resolver called for agent_id: {agent_id}, tenant_id: {tenant_id}") - 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 initialize_observability( - service_name: str = None, - service_namespace: str = None, -) -> bool: - """ - Initialize Agent 365 Observability SDK. - - Args: - service_name: Name of the service (defaults to OBSERVABILITY_SERVICE_NAME env var) - service_namespace: Namespace of the service (defaults to OBSERVABILITY_SERVICE_NAMESPACE env var) - - Returns: - True if configuration succeeded, False otherwise - """ - try: - status = configure( - service_name=service_name or os.getenv("OBSERVABILITY_SERVICE_NAME", "crewai-sample-agent"), - service_namespace=service_namespace or os.getenv("OBSERVABILITY_SERVICE_NAMESPACE", "agent365-samples"), - token_resolver=token_resolver, - ) - - if not status: - logger.warning("Agent 365 Observability configuration failed") - return False - - logger.info("Agent 365 Observability configured successfully") - return True - - except Exception as e: - logger.error(f"Error setting up observability: {e}") - return False diff --git a/python/crewai/sample_agent/pyproject.toml b/python/crewai/sample_agent/pyproject.toml index c645a874..2eca6b4c 100644 --- a/python/crewai/sample_agent/pyproject.toml +++ b/python/crewai/sample_agent/pyproject.toml @@ -1,23 +1,28 @@ [project] name = "crew_agent" version = "0.1.0" -description = "crew-agent using crewAI" -authors = [{ name = "Your Name", email = "you@example.com" }] +description = "CrewAI Sample Agent using Microsoft Agent 365 SDK with MCP tooling support" +authors = [{ name = "Microsoft", email = "support@microsoft.com" }] requires-python = ">=3.11" dependencies = [ - "crewai[tools]==1.4.1", + "crewai[azure-ai-inference,tools]==1.4.1", "tavily-python>=0.3.0", "python-dotenv>=1.0.0", + # Microsoft 365 Agents SDK - hosting and auth "microsoft-agents-hosting-aiohttp", "microsoft-agents-hosting-core", "microsoft-agents-authentication-msal", "microsoft-agents-activity", + # Core hosting deps "aiohttp", + # Microsoft Agent 365 SDK packages "microsoft_agents_a365_tooling >= 0.1.0", "microsoft_agents_a365_observability_core >= 0.1.0", + "microsoft-agents-a365-observability-hosting>=0.2.0", + "microsoft_agents_a365_notifications >= 0.1.0", "microsoft_agents_a365_runtime >= 0.1.0", "wrapt>=2.0.1", ] diff --git a/python/crewai/sample_agent/src/crew_agent/config/tasks.yaml b/python/crewai/sample_agent/src/crew_agent/config/tasks.yaml index 0e95f524..b16a93b2 100644 --- a/python/crewai/sample_agent/src/crew_agent/config/tasks.yaml +++ b/python/crewai/sample_agent/src/crew_agent/config/tasks.yaml @@ -7,9 +7,12 @@ weather_check_task: - Wind speed and direction - Humidity and visibility - Any weather warnings or adverse conditions + + IMPORTANT: Do NOT send any emails. Your role is ONLY to gather weather data. + The Driving Safety Specialist will handle all email communications. expected_output: > A detailed weather report for {location} with all relevant weather metrics - and conditions clearly formatted. + and conditions clearly formatted. Do NOT send emails - just provide the data. agent: weather_checker driving_safety_task: @@ -25,10 +28,17 @@ driving_safety_task: - Visibility: Poor visibility affects overall driving safety Provide a clear safety assessment with specific reasons based on the weather data. + + If the user requested an email, YOU are responsible for sending ONE email that includes: + 1. The complete weather report from the previous task + 2. Your driving safety assessment + Combine both into a single comprehensive email. Do NOT send multiple emails. + Use the SendEmailWithAttachmentsAsync tool to send the email directly (do NOT use draft tools). expected_output: > A clear safety assessment stating whether it's safe to drive with summer tires in the current weather conditions, along with specific reasons and any recommendations or warnings. Format as a clear decision with supporting details. + If an email was requested, confirm it was sent (only once, with combined content). agent: driving_safety_advisor context: - weather_check_task diff --git a/python/crewai/sample_agent/src/crew_agent/crew.py b/python/crewai/sample_agent/src/crew_agent/crew.py index 831d57b8..eaabba00 100644 --- a/python/crewai/sample_agent/src/crew_agent/crew.py +++ b/python/crewai/sample_agent/src/crew_agent/crew.py @@ -16,7 +16,14 @@ class CrewAgent(): tasks: List[Task] def __init__(self, mcps: Optional[list] = None): - self.mcps = mcps or [] + """ + Initialize CrewAgent with optional MCP tools. + + Args: + mcps: List of observable MCP tool wrappers (CrewAI BaseTool instances) + These are tools with ExecuteToolScope observability, not raw MCP configs. + """ + self.mcp_tools = mcps or [] # Learn more about YAML configuration files here: # Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended @@ -27,19 +34,21 @@ def __init__(self, mcps: Optional[list] = None): @agent def weather_checker(self) -> Agent: from crew_agent.tools.custom_tool import WeatherTool + # Combine built-in tools with observable MCP tools + all_tools = [WeatherTool()] + self.mcp_tools return Agent( config=self.agents_config['weather_checker'], # type: ignore[index] - tools=[WeatherTool()], + tools=all_tools, verbose=True, - mcps=self.mcps ) @agent def driving_safety_advisor(self) -> Agent: + # Pass observable MCP tools with ExecuteToolScope tracing return Agent( config=self.agents_config['driving_safety_advisor'], # type: ignore[index] + tools=self.mcp_tools, verbose=True, - mcps=self.mcps, ) # To learn more about structured task outputs, diff --git a/python/crewai/sample_agent/start_with_generic_host.py b/python/crewai/sample_agent/start_with_generic_host.py index b98d3fbe..3b2f0eac 100644 --- a/python/crewai/sample_agent/start_with_generic_host.py +++ b/python/crewai/sample_agent/start_with_generic_host.py @@ -3,9 +3,54 @@ #!/usr/bin/env python3 """ Example: Direct usage of Generic Agent Host with CrewAI wrapper. + +IMPORTANT: Observability must be configured BEFORE importing CrewAI +to prevent CrewAI from setting up its own TracerProvider. """ import sys +import os + +from constants import DEFAULT_SERVICE_NAME, DEFAULT_SERVICE_NAMESPACE + +# ============================================================ +# CRITICAL: Configure observability BEFORE importing CrewAI +# CrewAI sets up its own TracerProvider which would override ours +# ============================================================ +def _configure_observability_early(): + """Configure observability before any CrewAI imports.""" + from dotenv import load_dotenv + load_dotenv() + + enable_observability = os.getenv("ENABLE_OBSERVABILITY", "true").lower() in ("true", "1", "yes") + if not enable_observability: + print("ℹ️ Observability disabled (ENABLE_OBSERVABILITY=false)") + return + + try: + # Import and configure observability FIRST + from microsoft_agents_a365.observability.core.config import configure as configure_observability + from token_cache import get_cached_agentic_token + + service_name = os.getenv("OBSERVABILITY_SERVICE_NAME", DEFAULT_SERVICE_NAME) + service_namespace = os.getenv("OBSERVABILITY_SERVICE_NAMESPACE", DEFAULT_SERVICE_NAMESPACE) + + def token_resolver(agent_id: str, tenant_id: str) -> str | None: + """Resolve authentication token for observability exporter""" + return get_cached_agentic_token(tenant_id, agent_id) + + configure_observability( + service_name=service_name, + service_namespace=service_namespace, + token_resolver=token_resolver, + cluster_category=os.getenv("PYTHON_ENVIRONMENT", "development"), + ) + print("✅ Observability configured (before CrewAI import)") + except Exception as e: + print(f"⚠️ Failed to configure observability early: {e}") + +# Configure observability BEFORE importing CrewAI +_configure_observability_early() try: from agent import CrewAIAgent diff --git a/python/crewai/sample_agent/token_cache.py b/python/crewai/sample_agent/token_cache.py index c13a2f0b..e5d45776 100644 --- a/python/crewai/sample_agent/token_cache.py +++ b/python/crewai/sample_agent/token_cache.py @@ -1,30 +1,131 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ -Token caching utilities for Agent 365 Observability exporter authentication. +Token Cache for Observability + +Uses SDK's AgenticTokenCache for token generation components, plus a sync cache +for already-exchanged tokens. This is needed because configure_observability() +requires a sync token_resolver, but the SDK's get_observability_token() is async. + +Pattern: +1. Register observability via SDK's AgenticTokenCache (stores Authorization + TurnContext) +2. Exchange token asynchronously and cache the resulting string +3. token_resolver retrieves the cached string synchronously """ import logging +from threading import Lock + +from microsoft_agents_a365.observability.hosting.token_cache_helpers.agent_token_cache import ( + AgenticTokenCache, + AgenticTokenStruct, +) logger = logging.getLogger(__name__) -# Global token cache for Agent 365 Observability exporter -_agentic_token_cache = {} +# SDK's token cache for observability registration +_sdk_token_cache = AgenticTokenCache() + +# Sync cache for already-exchanged token strings +# Key format: "agent_id:tenant_id" (matching SDK's format) +_exchanged_tokens: dict[str, str] = {} +_lock = Lock() + + +def register_observability( + agent_id: str, + tenant_id: str, + token_generator: AgenticTokenStruct, + observability_scopes: list[str], +) -> None: + """ + Register observability using SDK's AgenticTokenCache. + + Args: + agent_id: Agent identifier + tenant_id: Tenant identifier + token_generator: AgenticTokenStruct with Authorization and TurnContext + observability_scopes: Scopes for token exchange + """ + _sdk_token_cache.register_observability( + agent_id=agent_id, + tenant_id=tenant_id, + token_generator=token_generator, + observability_scopes=observability_scopes, + ) + + +async def get_and_cache_observability_token(agent_id: str, tenant_id: str) -> str | None: + """ + Get observability token from SDK cache and cache the result for sync access. + + This is the bridge between async token exchange and sync token_resolver. + + Args: + agent_id: Agent identifier + tenant_id: Tenant identifier + + Returns: + Token if available, None otherwise + """ + # Try SDK's async token exchange + token = await _sdk_token_cache.get_observability_token(agent_id, tenant_id) + + if token: + # Cache for sync access by token_resolver + cache_key = f"{agent_id}:{tenant_id}" + with _lock: + _exchanged_tokens[cache_key] = token + logger.debug(f"Cached exchanged token for {cache_key}") + + return token def cache_agentic_token(tenant_id: str, agent_id: str, token: str) -> None: - """Cache the agentic token for use by Agent 365 Observability exporter.""" - key = f"{tenant_id}:{agent_id}" - _agentic_token_cache[key] = token - logger.debug(f"Cached agentic token for {key}") + """ + Cache an already-exchanged agentic token for sync access. + + Use this when you have the token from external exchange (e.g., BEARER_TOKEN env var). + + Args: + tenant_id: Tenant identifier + agent_id: Agent identifier + token: Already-exchanged agentic authentication token + """ + cache_key = f"{agent_id}:{tenant_id}" + with _lock: + _exchanged_tokens[cache_key] = token + logger.debug(f"Cached agentic token for {cache_key}") def get_cached_agentic_token(tenant_id: str, agent_id: str) -> str | None: - """Retrieve cached agentic token for Agent 365 Observability exporter.""" - key = f"{tenant_id}:{agent_id}" - token = _agentic_token_cache.get(key) + """ + Retrieve a cached agentic token synchronously. + + This is called by token_resolver in configure_observability(). + + Args: + tenant_id: Tenant identifier + agent_id: Agent identifier + + Returns: + Cached token if found, None otherwise + """ + cache_key = f"{agent_id}:{tenant_id}" + with _lock: + token = _exchanged_tokens.get(cache_key) + if token: - logger.debug(f"Retrieved cached agentic token for {key}") + logger.debug(f"Retrieved cached token for {cache_key}") else: - logger.debug(f"No cached token found for {key}") + logger.debug(f"No cached token found for {cache_key}") + return token + + +def clear_token_cache() -> None: + """Clear all cached tokens.""" + with _lock: + _exchanged_tokens.clear() + logger.debug("Token cache cleared") diff --git a/python/crewai/sample_agent/turn_context_utils.py b/python/crewai/sample_agent/turn_context_utils.py index f18dc589..c290e9bd 100644 --- a/python/crewai/sample_agent/turn_context_utils.py +++ b/python/crewai/sample_agent/turn_context_utils.py @@ -9,6 +9,7 @@ tenant information from the Microsoft Agents SDK TurnContext. """ +import os import uuid from dataclasses import dataclass from typing import Optional @@ -21,7 +22,11 @@ ExecutionType, InvokeAgentDetails, ) +from microsoft_agents_a365.observability.core.middleware.baggage_builder import BaggageBuilder from microsoft_agents_a365.observability.core.models.caller_details import CallerDetails +from microsoft_agents_a365.observability.hosting.scope_helpers.populate_baggage import populate + +from constants import DEFAULT_AGENT_ID @dataclass @@ -62,8 +67,10 @@ def extract_turn_context_details(context: TurnContext) -> TurnContextDetails: # Extract agent details from recipient (ChannelAccount) tenant_id = recipient.tenant_id if recipient else None agent_id = getattr(recipient, "id", None) if recipient else None + if not agent_id: + agent_id = os.getenv("AGENT_ID", DEFAULT_AGENT_ID) agent_name = getattr(recipient, "name", None) if recipient else None - agent_upn = getattr(recipient, "name", None) if recipient else None # ChannelAccount doesn't have UPN + agent_upn = getattr(recipient, "name", None) if recipient else None agent_blueprint_id = getattr(recipient, "agentic_app_id", None) if recipient else None agent_auid = getattr(recipient, "agentic_user_id", None) if recipient else None @@ -72,14 +79,21 @@ def extract_turn_context_details(context: TurnContext) -> TurnContextDetails: correlation_id = str(uuid.uuid4()) # Extract caller details from from_property (ChannelAccount) - # Note: ChannelAccount has: id, name, aad_object_id, role, agentic_user_id, agentic_app_id, tenant_id caller = activity.from_property if activity and activity.from_property else None caller_id = getattr(caller, "id", None) caller_name = getattr(caller, "name", None) caller_aad_object_id = getattr(caller, "aad_object_id", None) + # Warn if using fallback tenant_id to help identify configuration issues in production + if not tenant_id: + import logging + logging.getLogger(__name__).warning( + "tenant_id not found in TurnContext; using 'default-tenant'. " + "This may make it difficult to distinguish between deployments in telemetry." + ) + return TurnContextDetails( - tenant_id=tenant_id, + tenant_id=tenant_id or "default-tenant", agent_id=agent_id, agent_name=agent_name, agent_upn=agent_upn, @@ -93,7 +107,7 @@ def extract_turn_context_details(context: TurnContext) -> TurnContextDetails: ) -def create_agent_details(details: TurnContextDetails, description: str = "AI agent") -> AgentDetails: +def create_agent_details(details: TurnContextDetails, description: str = "AI agent powered by CrewAI framework") -> AgentDetails: """ Create AgentDetails from extracted TurnContextDetails. @@ -127,9 +141,9 @@ def create_caller_details(details: TurnContextDetails) -> CallerDetails: CallerDetails for observability """ return CallerDetails( - caller_id=details.caller_id, - caller_upn=details.caller_name, - caller_user_id=details.caller_aad_object_id or details.caller_id, + caller_id=details.caller_id or "unknown-caller", + caller_upn=details.caller_name or "unknown-user", + caller_user_id=details.caller_aad_object_id or details.caller_id or "unknown-user-id", caller_name=details.caller_name, ) @@ -165,7 +179,7 @@ def create_request(details: TurnContextDetails, message: str) -> Request: ) -def create_invoke_agent_details(details: TurnContextDetails, description: str = "AI agent") -> InvokeAgentDetails: +def create_invoke_agent_details(details: TurnContextDetails, description: str = "AI agent powered by CrewAI framework") -> InvokeAgentDetails: """ Create InvokeAgentDetails from extracted TurnContextDetails. @@ -181,3 +195,26 @@ def create_invoke_agent_details(details: TurnContextDetails, description: str = details=agent_details, session_id=details.conversation_id, ) + + +def build_baggage_builder(context: TurnContext, correlation_id: Optional[str] = None) -> BaggageBuilder: + """ + Build a BaggageBuilder populated from TurnContext activity. + + Uses the SDK's populate() function which extracts the following from TurnContext: + - tenant_id: From activity.conversation.tenant_id + - agent_id: From activity.recipient.id + - correlation_id: From activity channel data or generates new one + + Args: + context: The TurnContext from the Microsoft Agents SDK + correlation_id: Optional correlation id to override the one extracted by populate() + + Returns: + Populated BaggageBuilder instance ready to use with .build() context manager + """ + builder = BaggageBuilder() + populate(builder, context) + if correlation_id: + builder.correlation_id(correlation_id) + return builder