Python: [Issue]: pydantic_core._pydantic_core.ValidationError with Anthropic Model

[udbj]

Language

Python

Type of Issue

Bug

agent-framework-core

1.0.0b251223

agent-framework-azure-ai

1.0.0b251223

agent-framework-anthropic

1.0.0b251223

agent-framework-azurefunctions

1.24.0

Description

Running a Group chat with three Anthropic agents. When the workflow ends, it throws the following exception:

pydantic_core._pydantic_core.ValidationError: 1 validation error for ManagerSelectionResponse
  Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value="Good! The file doesn't ...ation check guidelines.", input_type=str]
    For further information visit https://errors.pydantic.dev/2.12/v/json_invalid

When I run the application using GPT-5.2, the streamed messages unpacked from AgentRunUpdateEvent are always in JSON format, for example:

[groupchat_agent:Orchestrator]: {"selected_participant":null,"instruction":null,"finish":true,"final_message":"Pre-execution check: Table already exists on drive.\n\nPer instructions, marking file as done (no execution needed)."}

However, event data from claude (sonnet 4.5) doesn't seem to packed as JSON like that, it's always plain text. Would it be possible for the framework to wrap everything in valid JSONs?

[eavanvalkenburg]

I'm not sure where your issue is coming from, could you share some sample code that demonstrates this, it might be in the anthropic clients, but it might also be somewhere in workflows. CC: @moonbox3

[udbj]

I'm not sure where your issue is coming from, could you share some sample code that demonstrates this, it might be in the anthropic clients, but it might also be somewhere in workflows. CC: @moonbox3

Hi Eduard, thanks for getting back to me.

Here's the main chunk of my script with some stuff obscured. It's pretty much the same as the sample code for GroupChatBuilder on the docs:

    #chat_client = OpenAIChatClient(model_id=OPENAI_CHAT_MODEL_ID)
    chat_client = AnthropicClient(model_id=ANTHROPIC_CHAT_MODEL_ID)
        
    # Create agents
        agent_01 = chat_client.create_agent(
        instructions=instructions_01,
        name="<name>",
        description="<desc>",
        tools=tool_01
    )

    agent_02 = chat_client.create_agent(
        instructions=instructions_02,
        name="<name>",
        description="<desc>",
        tools=tool_02
    )

    orchestrator_agent = chat_client.create_agent(
        instructions=orchestrator_instructions,
        name="<name>",
        description="<desc>",
        tools=[tool_04, tool_05, tool_06]
    )

    workflow = (
        GroupChatBuilder()
        .set_manager(orchestrator_agent, display_name="Orchestrator")
        .participants([agent_01, agent_02])
        .build()
    )

    file_checklist = {}


    for script_file in SCRIPT_SRC_PATH.rglob("*.py"):
        file_checklist[script_file.name] = script_file.resolve()

    global progress_record

    while len(file_checklist) > 0:

        for file_name, file_path in file_checklist.items():

            with open(file_path) as f:
                file_text = f.read()
            
            input_msg = \
            f"""
            file: {file_name}
            {"-" * 60}
            ```py
            {file_text}
            ```
            """

            final_conversation: list[ChatMessage] = []
            last_executor_id: str | None = None
            try:
                # Run the workflow
                async for event in workflow.run_stream(input_msg):
                    if isinstance(event, AgentRunUpdateEvent):
                        # Print streaming agent updates
                        eid = event.executor_id
                        if eid != last_executor_id:
                            if last_executor_id is not None:
                                print()
                            print(f"[{eid}]:", end=" ", flush=True)
                            last_executor_id = eid
                        print(event.data, end="", flush=True)
                    elif isinstance(event, WorkflowOutputEvent):
                        # Workflow completed - data is a list of ChatMessage
                        final_conversation = cast(list[ChatMessage], event.data)
                
                if final_conversation:
                    print("\n\n" + "=" * 80)
                    print("Final Conversation:")
                    for msg in final_conversation:
                        author = getattr(msg, "author_name", "Unknown")
                        text = getattr(msg, "text", str(msg))
                        print(f"\n[{author}]\n{text}")
                        print("-" * 80)

            except HTTPError as e:
                # for rate limits
                tb.format_exc(e)
                time.sleep(60)
            
        for done_file in progress_record:
            if done_file in file_checklist:
                file_checklist.pop(done_file)

    print("\nAll Files Processed!")

(The loop seems to be infinite but progress_record is updated by the agents through tools so it ends when all files are processed)

Again - works perfectly fine with GPT, but none of the output from Claude is in JSONs.