Skip to content

Python: [Bug]: PydanticSchemaGenerationError in FunctionInvocationContext #4809

@ashishmundra4001

Description

@ashishmundra4001

Description

The examples here shows using FunctionInvocationContext in @tool - https://learn.microsoft.com/en-us/agent-framework/agents/middleware/runtime-context?pivots=programming-language-python

So I tried to use it in my agent pasted in code sample below but I get a Pydantic Error. If code looks a lot, just see get_weather function in it which is used as a @tool.

If I define ctx without a type then my agent can run, but ctx just gets a string which has some summary of query I ask like "get weather information" whereas I am interested to read ctx.kwargs that I have set in my middlewares so I need to probably define ctx as FunctionInvocationContext type to be able to then fetch kwargs.

This agent is a Foundry Hosted Agent and is run using from_agent_framework(agent).run()

The /responses api body is like:
{
"agent": {
"name": "Super-Agent",
"type": "agent_reference"
},
"stream": false,
"input": "What is the weather in San francisco.",
"metadata": {
"x-some-token": "some token here",
"x-dev-authorization-token": "dev env only - caller's oauth token here"
}
}

My intention was to fetch metadata in AgentMiddleware and then use it in my tool via FunctionInvocationContext.

Code Sample

"""Agent factory for creating and configuring agents."""
import logging
from typing import List, Dict

from agent_framework.azure import AzureOpenAIChatClient
from agent_framework import FunctionInvocationContext, tool
from random import randint
from middleware import UserContextAgentMiddleware, UserContextFunctionMiddleware
from config import AppConfig
from .mcp_manager import MCPToolManager
from auth.token_service import TokenService

logger = logging.getLogger(__name__)

class AgentFactory:
    """Factory class for creating and configuring agents."""
    
    BASE_INSTRUCTIONS = "You are a helpful assistant with access to MCP tools. Always answer using appropriate tools."
    AGENT_NAME = "SuperAgent"
    
    def __init__(self, config: AppConfig):
        """
        Initialize agent factory.
        
        Args:
            config: Application configuration
        """
        self.config = config
        self.token_service = None
        self.chat_client = None
        self.mcp_manager = None
        
    def setup_credentials(self) -> None:
        """Set up Azure credentials and token service for agent.""" 
        
        # Create token service for agent authentication
        self.token_service = TokenService(self.config)
        
        logger.info("Agent credentials and token service initialized")
        
    def setup_chat_client(self) -> None:
        """Create and configure Azure OpenAI chat client for agent."""
        if not self.token_service:
            raise RuntimeError("Must call setup_credentials() before setup_chat_client()")
            
        # Use the token service's cognitive services token provider for chat client
        self.chat_client = AzureOpenAIChatClient(
            credential=self.token_service.cognitive_services_token_provider,
            endpoint=self.config.azure_openai_endpoint,
            deployment_name=self.config.azure_openai_chat_deployment_name
        )
        
        logger.info("Agent chat client initialized")
        
    def build_instructions(self, tool_info: List[Dict[str, str]]) -> str:
        """Build dynamic instructions with available tools information."""
        instructions = self.BASE_INSTRUCTIONS
        
        if tool_info:
            tools_description = "\n\nAvailable tools:\n"
            for tool in tool_info:
                tools_description += f"- {tool['name']}: {tool['description']}\n"
            instructions = instructions + tools_description
            
        logger.debug(f"Final instructions: {instructions}")
        
        return instructions

    async def create_agent(self):
        """
        Create agent with MCP tools and dynamic instructions.
        """
        # Setup agent credentials and services
        self.setup_credentials()
        self.setup_chat_client() 
        
        # Create MCP tool manager
        self.mcp_manager = MCPToolManager(
            config=self.config
        )
        
        # Create and configure MCP tool
        mcp_tool = self.mcp_manager.create_mcp_tool()
        
        # Get tool information
        # ToDo: Get Tool information to build instructions - currently there is some error to load tools
        # tool_info = await self.mcp_manager.get_tool_info()
        tool_info = []
        
        # Log tool information  
        if tool_info:
            logger.debug(f"Available MCP tools ({len(tool_info)}):")
            for i, tool in enumerate(tool_info, 1):
                logger.debug(f"   {i}. {tool['name']}: {tool['description']}")
        else:
            logger.debug("MCP tool configured - tools will be discovered on first request")
        
        # Build instructions with tool info
        instructions = self.build_instructions(tool_info)
        
        logger.info(f"Creating agent: {self.AGENT_NAME}")
        
        agent = self.chat_client.as_agent(
            name=self.AGENT_NAME,
            instructions=instructions,
            tools=[mcp_tool,get_weather],
            middleware=[UserContextAgentMiddleware(), UserContextFunctionMiddleware()]
        )
        
        logger.info("Agent created successfully")
        return agent
        
@tool(approval_mode="never_require")
def get_weather(
    location: str,
    ctx: FunctionInvocationContext,
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    print(f"get_weather called with location: {location} and kwargs: {ctx.kwargs}, metadata: {ctx.metadata}, arguments: {ctx.arguments} ")
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."

Error Messages / Stack Traces

pydantic.errors.PydanticSchemaGenerationError: Unable to generate pydantic-core schema for <class 'agent_framework._middleware.FunctionInvocationContext'>. Set `arbitrary_types_allowed=True` in the model_config to ignore this error or implement `__get_pydantic_core_schema__` on your type to fully support it.

If you got this error by calling handler(<some type>) within `__get_pydantic_core_schema__` then you likely need to call `handler.generate_schema(<some type>)` since we do not call `__get_pydantic_core_schema__` on `<some type>` otherwise to avoid infinite recursion.

Package Versions

azure-ai-agentserver-agentframework==1.0.0b17 agent-framework-azure-ai

Python Version

3.12

Additional Context

No response

Metadata

Metadata

Assignees

Labels

bugSomething isn't workingpython

Type

Projects

Status

Planned

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions