Agent Intelligence System

class GraphBuilder:
    def buildgraph(self):
        agent_node = _create_agent_node(self.tools)

        workflow = StateGraph(AgentState)
        workflow.add_node("agent", agent_node)
        workflow.add_node("tool_execution", ToolNode(self.tools))
        workflow.add_edge(START, "agent")
        workflow.add_conditional_edges(
            "agent",
            tools_condition,
            {
                "tools": "tool_execution",
                END: END,
            },
        )
        workflow.add_edge("tool_execution", "agent")
        return workflow.compile()

def _create_agent_node(tools: Sequence[StructuredTool]):
    bound_llm = _llm.bind_tools(list(tools))

    async def _agent_node(state: AgentState, **_):
        messages = list(state["messages"])
        if not messages or not isinstance(messages[0], SystemMessage):
            messages = [_system_message] + messages
        response = await bound_llm.ainvoke(messages)
        return {"messages": [response]}

    return _agent_node

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

class GitHubToolInput(BaseModel):
    url: HttpUrl = Field(..., description="Full URL to a public GitHub repository.")
    question: str = Field(..., description="Question about the repository.")
    chat_history: Optional[list[dict[str, Any]]] = Field(
        default=None,
        description="Optional chat history as a list of {role, content} maps.",
    )

class GmailSendEmailInput(BaseModel):
    to: EmailStr = Field(..., description="Recipient email address.")
    subject: str = Field(..., min_length=1, description="Email subject line.")
    body: str = Field(..., min_length=1, description="Plain-text body content.")
    access_token: Optional[str] = Field(
        default=None,
        description="OAuth access token with Gmail send scope. "
                   "If omitted, a pre-configured token will be used when available.",
    )

def build_agent_tools(context: Optional[Dict[str, Any]] = None) -> list[StructuredTool]:
    ctx: Dict[str, Any] = dict(context or {})
    google_token = ctx.get("google_access_token") or ctx.get("google_acces_token")
    pyjiit_payload = ctx.get("pyjiit_login_response") or ctx.get("pyjiit_login_responce")

    tools: list[StructuredTool] = [
        github_agent,
        websearch_agent,
        website_agent,
        youtube_agent,
        browser_action_agent,
    ]

    if google_token:
        tools.append(
            StructuredTool(
                name=gmail_agent.name,
                description=gmail_agent.description,
                coroutine=partial(_gmail_tool, _default_token=google_token),
                args_schema=GmailToolInput,
            )
        )
        # ... similar pattern for other Google tools

    if pyjiit_payload:
        tools.append(
            StructuredTool(
                name=pyjiit_agent.name,
                description=pyjiit_agent.description,
                coroutine=partial(_pyjiit_attendance_tool, _default_payload=pyjiit_payload),
                args_schema=PyjiitAttendanceInput,
            )
        )

    return tools

async def _gmail_tool(
    access_token: Optional[str] = None,
    max_results: int = 5,
    *,
    _default_token: Optional[str] = None,
) -> str:
    token = access_token or _default_token
    if not token:
        return "Unable to fetch Gmail messages because no Google access token was provided."
    # ... rest of implementation

async def _github_tool(url: HttpUrl, question: str, chat_history: Optional[list[dict]] = None) -> str:
    repo_data = await convert_github_repo_to_markdown(url)
    history = _format_chat_history(chat_history)
    payload = {
        "question": question,
        "text": repo_data.content,
        "tree": repo_data.tree,
        "summary": repo_data.summary,
        "chat_history": history,
    }
    response = await asyncio.to_thread(github_chain.invoke, payload)
    return _ensure_text(response)

async def _websearch_tool(query: str, max_results: int = 5) -> str:
    bounded = max(1, min(10, max_results))
    results = await asyncio.to_thread(web_search_pipeline, query, None, bounded)
    if not results:
        return "No web results were found."

    snippets: list[str] = []
    for item in results[:bounded]:
        url = item.get("url", "")
        text = (item.get("md_body_content") or "").strip().replace("\n", " ")
        if len(text) > 320:
            text = text[:320].rstrip() + "..."
        snippets.append(f"URL: {url}\nSummary: {text}")

    return "\n\n".join(snippets)

async def _website_tool(url: HttpUrl, question: str, chat_history: Optional[list[dict]] = None) -> str:
    markdown = await asyncio.to_thread(markdown_fetcher, str(url))
    history = _format_chat_history(chat_history)
    response = await asyncio.to_thread(
        get_website_answer,
        website_chain,
        question,
        markdown,
        history,
    )
    return _ensure_text(response)

async def _youtube_tool(url: HttpUrl, question: str, chat_history: Optional[list[dict]] = None) -> str:
    history = _format_chat_history(chat_history)
    response = await asyncio.to_thread(
        get_youtube_answer,
        youtube_chain,
        question,
        str(url),
        history,
    )
    return _ensure_text(response)

async def _gmail_tool(
    access_token: Optional[str] = None,
    max_results: int = 5,
    *,
    _default_token: Optional[str] = None,
) -> str:
    token = access_token or _default_token
    if not token:
        return "Unable to fetch Gmail messages because no Google access token was provided."

    bounded = max(1, min(25, max_results))

    try:
        messages = await asyncio.to_thread(get_latest_emails, token, max_results=bounded)
        return _ensure_text({"messages": messages})
    except Exception as exc:
        return f"Failed to fetch Gmail messages: {exc}"

async def _pyjiit_attendance_tool(
    registration_code: Optional[str] = None,
    session_payload: Optional[Dict[str, Any]] = None,
    *,
    _default_payload: Optional[Dict[str, Any]] = None,
) -> str:
    payload = session_payload or _default_payload
    if not payload:
        return "Unable to fetch attendance because no PyJIIT login session was provided."

    # ... session construction and attendance fetching logic

_llm = LargeLanguageModel().client

DEFAULT_SYSTEM_PROMPT = (
    "You are a helpful AI assistant that maintains conversation context and "
    "remembers useful information shared by users. Use the available tools "
    "when they can improve the answer, otherwise reply directly. "
    "Credentials such as Google access tokens and PyJIIT login sessions are provided "
    "automatically; never request them from the user. If a request involves JIIT "
    "attendance or portal data, call the 'pyjiit_agent' tool immediately using the "
    "existing session. If that session fails, report that the login session expired "
    "and ask the user to refresh it via the secure flow—do not ask for usernames or passwords."
)

if not messages or not isinstance(messages[0], SystemMessage):
    messages = [_system_message] + messages

try:
    messages = await asyncio.to_thread(get_latest_emails, token, max_results=bounded)
    return _ensure_text({"messages": messages})
except Exception as exc:
    return f"Failed to fetch Gmail messages: {exc}"

except Exception as exc:
    logger.error("Error generating react agent answer: %s", exc)
    return (
        "I apologize, but I encountered an error processing your question. "
        "Please try again."
    )