fix pylint

Author: lusu-msft

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/langgraph.py

@@ -1,17 +1,18 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+# pylint: disable=logging-fstring-interpolation,broad-exception-caught
 import os
 import re
 from typing import Optional
 
+from langchain_core.runnables import RunnableConfig
+from langgraph.graph.state import CompiledStateGraph
+
 from azure.ai.agentserver.core.constants import Constants
 from azure.ai.agentserver.core.logger import get_logger
 from azure.ai.agentserver.core.server.base import FoundryCBAgent
 from azure.ai.agentserver.core.server.common.agent_run_context import AgentRunContext
-from langchain_core.runnables import RunnableConfig
-
-from langgraph.graph.state import CompiledStateGraph
 
 from .models import (
     LanggraphMessageStateConverter,
@@ -31,9 +32,10 @@ def __init__(self, graph: CompiledStateGraph, state_converter: Optional[Langgrap
         """
         Initialize the LangGraphAdapter with a CompiledStateGraph.
 
-        Args:
-            graph (StateGraph): The LangGraph StateGraph to adapt.
-            state_converter: custom state converter. Required if graph state is not MessagesState.
+        :param graph: The LangGraph StateGraph to adapt.
+        :type graph: CompiledStateGraph
+        :param state_converter: custom state converter. Required if graph state is not MessagesState.
+        :type state_converter: Optional[LanggraphStateConverter]
         """
         super().__init__()
         self.graph = graph
@@ -52,8 +54,7 @@ async def agent_run(self, context: AgentRunContext):
         if not context.stream:
             response = await self.agent_run_non_stream(input_data, context)
             return response
-        else:
-            return self.agent_run_astream(input_data, context)
+        return self.agent_run_astream(input_data, context)
 
     def init_tracing_internal(self, exporter_endpoint=None, app_insights_conn_str=None):
         # set env vars for langsmith
@@ -87,11 +88,13 @@ async def agent_run_non_stream(self, input_data: dict, context: AgentRunContext)
         """
         Run the agent with non-streaming response.
 
-        Args:
-            context (AgentRunContext): The context for the agent run.
+        :param input_data: The input data to run the agent with.
+        :type input_data: dict
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
 
-        Returns:
-            RunObject: The result of the agent run.
+        :return: The response of the agent run.
+        :rtype: dict
         """
 
         try:
@@ -108,11 +111,13 @@ async def agent_run_astream(self, input_data: dict, context: AgentRunContext):
         """
         Run the agent with streaming response.
 
-        Args:
-            request_body (CreateResponse): The request body to run the agent with.
+        :param input_data: The input data to run the agent with.
+        :type input_data: dict
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
 
-        Returns:
-            StreamingResponse: The streaming response of the agent run.
+        :return: An async generator yielding the response stream events.
+        :rtype: AsyncGenerator[dict]
         """
         try:
             logger.info(f"Starting streaming agent run {context.response_id}")
@@ -128,6 +133,12 @@ async def agent_run_astream(self, input_data: dict, context: AgentRunContext):
     def create_runnable_config(self, context: AgentRunContext) -> RunnableConfig:
         """
         Create a RunnableConfig from the converted request data.
+
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
+
+        :return: The RunnableConfig for the agent run.
+        :rtype: RunnableConfig
         """
         config = RunnableConfig(
             configurable={

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/langgraph_request_converter.py

@@ -1,11 +1,10 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+# pylint: disable=logging-fstring-interpolation
 import json
 from typing import Dict, List
 
-from azure.ai.agentserver.core.logger import get_logger
-from azure.ai.agentserver.core.models import CreateResponse, openai as openai_models, projects as project_models
 from langchain_core.messages import (
     AIMessage,
     AnyMessage,
@@ -15,6 +14,9 @@
 )
 from langchain_core.messages.tool import ToolCall
 
+from azure.ai.agentserver.core.logger import get_logger
+from azure.ai.agentserver.core.models import CreateResponse, openai as openai_models, projects as project_models
+
 logger = get_logger()
 
 role_mapping = {
@@ -61,29 +63,40 @@ def convert(self) -> dict:
     def convert_input(self, item: openai_models.ResponseInputItemParam) -> AnyMessage:
         """
         Convert ResponseInputItemParam to a LangGraph message
+
+        :param item: The ResponseInputItemParam to convert from request.
+        :type item: openai_models.ResponseInputItemParam
+
+        :return: The converted LangGraph message.
+        :rtype: AnyMessage
         """
         item_type = item.get("type", project_models.ItemType.MESSAGE)
         if item_type == project_models.ItemType.MESSAGE:
             # this is a message
             return self.convert_message(item)
-        elif item_type == project_models.ItemType.FUNCTION_CALL:
+        if item_type == project_models.ItemType.FUNCTION_CALL:
             return self.convert_function_call(item)
-        elif item_type == project_models.ItemType.FUNCTION_CALL_OUTPUT:
+        if item_type == project_models.ItemType.FUNCTION_CALL_OUTPUT:
             return self.convert_function_call_output(item)
-        else:
-            raise ValueError(f"Unsupported OpenAIItemParam type: {item_type}, {item}")
+        raise ValueError(f"Unsupported OpenAIItemParam type: {item_type}, {item}")
 
     def convert_message(self, message: dict) -> AnyMessage:
         """
         Convert a message dict to a LangGraph message
+
+        :param message: The message dict to convert.
+        :type message: dict
+
+        :return: The converted LangGraph message.
+        :rtype: AnyMessage
         """
         content = message.get("content")
         role = message.get("role", project_models.ResponsesMessageRole.USER)
         if not content:
             raise ValueError(f"Message missing content: {message}")
         if isinstance(content, str):
             return role_mapping[role](content=content)
-        elif isinstance(content, list):
+        if isinstance(content, list):
             return role_mapping[role](content=self.convert_OpenAIItemContentList(content))
         raise ValueError(f"Unsupported ResponseMessagesItemParam content type: {type(content)}, {content}")
 
@@ -100,20 +113,26 @@ def convert_function_call(self, item: dict) -> AnyMessage:
 
     def convert_function_call_output(self, item: dict) -> ToolMessage:
         try:
-            item = openai_models.response_input_item_param.FunctionCallOutput(**item)
+            item = openai_models.response_input_item_param.FunctionCallOutput(**item) # pylint: disable=no-member
         except Exception as e:
             raise ValueError(f"Invalid function call output item: {item}") from e
 
         output = item.get("output", None)
         if isinstance(output, str):
             return ToolMessage(content=output, tool_call_id=item.get("call_id"))
-        elif isinstance(output, list):
+        if isinstance(output, list):
             return ToolMessage(content=self.convert_OpenAIItemContentList(output), tool_call_id=item.get("call_id"))
         raise ValueError(f"Unsupported function call output type: {type(output)}, {output}")
 
-    def convert_OpenAIItemContentList(self, content: List[Dict]) -> List:
+    def convert_OpenAIItemContentList(self, content: List[Dict]) -> List[Dict]:
         """
         Convert ItemContent to a list format
+
+        :param content: The list of ItemContent to convert.
+        :type content: List[Dict]
+
+        :return: The converted list of ItemContent.
+        :rtype: List[Dict]
         """
         result = []
         for item in content:
@@ -123,6 +142,12 @@ def convert_OpenAIItemContentList(self, content: List[Dict]) -> List:
     def convert_OpenAIItemContent(self, content: Dict) -> Dict:
         """
         Convert ItemContent to a dict format
+
+        :param content: The ItemContent to convert.
+        :type content: Dict
+
+        :return: The converted ItemContent.
+        :rtype: Dict
         """
         res = content.copy()
         content_type = content.get("type")

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/langgraph_response_converter.py

@@ -1,14 +1,16 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+# pylint: disable=logging-fstring-interpolation,broad-exception-caught,logging-not-lazy
 import copy
 from typing import List
 
+from langchain_core import messages
+from langchain_core.messages import AnyMessage
+
 from azure.ai.agentserver.core.logger import get_logger
 from azure.ai.agentserver.core.models import projects as project_models
 from azure.ai.agentserver.core.server.common.agent_run_context import AgentRunContext
-from langchain_core import messages
-from langchain_core.messages import AnyMessage
 
 from .utils import extract_function_call
 
@@ -24,19 +26,19 @@ def convert(self) -> project_models.ItemResource:
         res = []
         for step in self.output:
             for node_name, node_output in step.items():
-                messages = node_output.get("messages")
-                if not messages:
+                message_arr = node_output.get("messages")
+                if not message_arr:
                     logger.warning(f"No messages found in node {node_name} output: {node_output}")
                     continue
-                for message in messages:
+                for message in message_arr:
                     try:
                         converted = self.convert_output_message(message)
                         res.append(converted)
                     except Exception as e:
                         logger.error(f"Error converting message {message}: {e}")
         return res
 
-    def convert_output_message(self, output_message: AnyMessage) -> project_models.ItemResource:
+    def convert_output_message(self, output_message: AnyMessage):  # pylint: disable=inconsistent-return-statements
         # Implement the conversion logic for inner inputs
         if isinstance(output_message, messages.HumanMessage):
             return project_models.ResponsesUserMessageItemResource(
@@ -86,14 +88,14 @@ def convert_output_message(self, output_message: AnyMessage) -> project_models.I
             )
 
     def convert_MessageContent(
-        self, content: str | list[str | dict], role: project_models.ResponsesMessageRole
+        self, content, role: project_models.ResponsesMessageRole
     ) -> List[project_models.ItemContent]:
         if isinstance(content, str):
             return [self.convert_MessageContentItem(content, role)]
         return [self.convert_MessageContentItem(item, role) for item in content]
 
     def convert_MessageContentItem(
-        self, content: str | dict, role: project_models.ResponsesMessageRole
+        self, content, role: project_models.ResponsesMessageRole
     ) -> project_models.ItemContent:
         content_dict = copy.deepcopy(content) if isinstance(content, dict) else {"text": content}
 

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/langgraph_state_converter.py

@@ -36,16 +36,19 @@
 class LanggraphStateConverter(ABC):
     """Abstract base class for LangGraph state <-> response conversion."""
 
-    def __init__(self):
-        super().__init__()
-
     @abstractmethod
     def get_stream_mode(self, context: AgentRunContext) -> str:
         """Return a string indicating streaming mode for this run.
 
         Examples: "values", "updates", "messages", "custom", "debug".
         Implementations may inspect context.request.stream or other flags.
         Must be fast and side-effect free.
+
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
+
+        :return: The streaming mode as a string.
+        :rtype: str
         """
 
     @abstractmethod
@@ -54,6 +57,12 @@ def request_to_state(self, context: AgentRunContext) -> Dict[str, Any]:
 
         Return a serializable dict that downstream graph execution expects.
         Should not mutate the context. Raise ValueError on invalid input.
+
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
+
+        :return: The initial LangGraph state as a dictionary.
+        :rtype: Dict[str, Any]
         """
 
     @abstractmethod
@@ -63,6 +72,14 @@ def state_to_response(self, state: Any, context: AgentRunContext) -> Response:
         Implementations must construct and return an models.Response.
         The returned object should include output items, usage (if available),
         and reference the agent / conversation from context.
+
+        :param state: The completed LangGraph state.
+        :type state: Any
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
+
+        :return: The final non-streaming Response object.
+        :rtype: Response
         """
 
     @abstractmethod
@@ -74,6 +91,14 @@ async def state_to_response_stream(
         Yield ResponseStreamEvent objects in the correct order. Implementations
         are responsible for emitting lifecycle events (created, in_progress, deltas,
         completed, errors) consistent with the OpenAI Responses streaming contract.
+
+        :param stream_state: An async iterator of partial LangGraph state updates.
+        :type stream_state: AsyncIterator[Dict[str, Any] | Any]
+        :param context: The context for the agent run.
+        :type context: AgentRunContext
+
+        :return: An async generator yielding ResponseStreamEvent objects.
+        :rtype: AsyncGenerator[ResponseStreamEvent, None]
         """
 
 
@@ -83,8 +108,7 @@ class LanggraphMessageStateConverter(LanggraphStateConverter):
     def get_stream_mode(self, context: AgentRunContext) -> str:
         if context.request.get("stream"):
             return "messages"
-        else:
-            return "updates"
+        return "updates"
 
     def request_to_state(self, context: AgentRunContext) -> Dict[str, Any]:
         converter = LangGraphRequestConverter(context.request)

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/langgraph_stream_response_converter.py

@@ -1,12 +1,14 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+# pylint: disable=logging-fstring-interpolation
 from typing import List
 
+from langchain_core.messages import AnyMessage
+
 from azure.ai.agentserver.core.logger import get_logger
 from azure.ai.agentserver.core.models import ResponseStreamEvent
 from azure.ai.agentserver.core.server.common.agent_run_context import AgentRunContext
-from langchain_core.messages import AnyMessage
 
 from .response_event_generators.response_event_generator import ResponseEventGenerator, StreamEventState
 from .response_event_generators.response_stream_event_generator import ResponseStreamEventGenerator

sdk/ai/azure-ai-agentserver-langgraph/azure/ai/agentserver/langgraph/models/response_event_generators/response_content_part_event_generator.py

@@ -1,11 +1,13 @@
 # ---------------------------------------------------------
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
+# pylint: disable=unused-argument,consider-using-in,consider-merging-isinstance
 from typing import List
 
-from azure.ai.agentserver.core.models import projects as project_models
 from langchain_core import messages as langgraph_messages
 
+from azure.ai.agentserver.core.models import projects as project_models
+
 from . import item_content_helpers
 from .response_event_generator import ResponseEventGenerator, StreamEventState
 from .response_output_text_event_generator import ResponseOutputTextEventGenerator
@@ -30,7 +32,7 @@ def __init__(
         self.item_content_helper = None
 
     def try_process_message(
-        self, message, run_details, stream_state: StreamEventState
+        self, message, context, stream_state: StreamEventState
     ) -> tuple[bool, ResponseEventGenerator, List[project_models.ResponseStreamEvent]]:
         is_processed = False
         events = []
@@ -41,14 +43,14 @@ def try_process_message(
                 self.logger.warning(f"Cannot create item content helper for message: {message}")
                 return True, self, []
         if self.item_content_helper and not self.started:
-            self.started, start_events = self.on_start(message, run_details, stream_state)
+            self.started, start_events = self.on_start(message, context, stream_state)
             if not self.started:
                 # could not start processing, skip this message
                 return True, self, []
             events.extend(start_events)
 
         if self.should_end(message):
-            complete_events = self.on_end(message, run_details, stream_state)
+            complete_events = self.on_end(message, context, stream_state)
             events.extend(complete_events)
             next_processor = self.parent
             is_processed = self.has_finish_reason(message) if message else False