【OpenAI Agents SDK＆MCP】自律するAIエージェントを試作 #LLM – Qiita

# 環境変数の読み込み
load_dotenv()
env_vars = os.environ.copy()

# ログインユーザーのメンバーコード（ログインユーザーもボドゲ部のメンバーの1人）
user_code_member = ""

# RunResultクラスのto_input_list()の結果を保持
conversation_items = []  # OpenAI Agents SDKの履歴管理を活用

# MCPサーバー「アナリスト・コンサルタント専用」
server_sug = None
# MCPサーバー「アナリスト・コンサルタント専用」の設定
SERVER_COMMAND_SUG = "uv"
SERVER_SCRIPT_SUG = "./mcp_servers/server_sug.py"

# MCPサーバー「処理・問い合わせ対応係専用」
server_reg = None
# MCPサーバー「処理・問い合わせ対応係専用」の設定
SERVER_COMMAND_REG = "uv"
SERVER_SCRIPT_REG = "./mcp_servers/server_reg.py"

# 全エージェント内部で使用するLLM
LLM_MODEL_NAME ="gpt-4o-mini"


def initialize(user_code: str) -> bool:
    """
    MCPクライアントの初期化。
    """
    
    global user_code_member, conversation_items, server_sug, server_reg

    # ログインユーザー（プロンプト投入者）のメンバーコード
    user_code_member = user_code
    
    # 会話履歴をクリア
    conversation_items = []

    # OpenAI APIキーの取得と設定
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEYが設定されていません。")

    set_default_openai_key(api_key)

    # MCPサーバー「アナリスト・コンサルタント専用」インスタンス生成
    server_sug = MCPServerStdio(
        params={
            "command": SERVER_COMMAND_SUG,
            "args": ["run", SERVER_SCRIPT_SUG],
            "env": env_vars
        }
    )

    # MCPサーバー「処理・問い合わせ対応係専用」インスタンス生成
    server_reg = MCPServerStdio(
        params={
            "command": SERVER_COMMAND_REG,
            "args": ["run", SERVER_SCRIPT_REG],
            "env": env_vars
        }
    )    
    
    return True
    

def process_user_prompt(user_input: str) -> str:
    """
    ユーザーのプロンプトuser_inputをAIエージェントが処理。
    AIエージェントによる最終応答を返す。
    UIから呼ばれることを想定し、同期関数とする。
    """

    # 子エージェント「アナリスト・コンサルタント」のinstructions
    instructions_sug = create_instructions_for_sug()

    # 子エージェント「処理・問い合わせ対応係」のinstructions
    instructions_reg = create_instructions_for_reg() 

    # 親エージェント「1次受付」のinstructions
    instructions_main = create_instructions_for_main() 

    # 子エージェント「アナリスト・コンサルタント」のhandoff_description
    handoff_description_sug = create_handoff_description_for_sug()

    # 子エージェント「処理・問い合わせ対応係」のhandoff_description
    handoff_description_reg = create_handoff_description_for_reg()
    

    async def async_process():
        global conversation_items, server_sug, server_reg

        # エージェントの最終応答で、この同期関数の戻り値
        ret = ""

        try:
 
            async with server_sug, server_reg:
                # async with ブロックに入り、MCPサーバーとのMCPセッションが自動的に開始される

                await server_sug.list_tools()
                await server_reg.list_tools()

                # 子エージェント「アナリスト・コンサルタント」インスタンス生成
                agent_sug = Agent(
                    name="アナリスト・コンサルタント",
                    instructions=instructions_sug,
                    mcp_servers=[server_sug],
                    model=LLM_MODEL_NAME,
                    handoff_description=handoff_description_sug
                )

                # 子エージェント「処理・問い合わせ対応係」インスタンス生成
                agent_reg = Agent(
                    name="処理・問い合わせ対応係",
                    instructions=instructions_reg,
                    mcp_servers=[server_reg],
                    model=LLM_MODEL_NAME,
                    handoff_description=handoff_description_reg
                )

                # 子エージェント同士で委譲しあえるように
                agent_sug.handoffs = [agent_reg]
                agent_reg.handoffs = [agent_sug]

                # 親エージェント「1次受付」インスタンス生成
                agent_main = Agent(
                    name="1次受付",
                    instructions=instructions_main,
                    model=LLM_MODEL_NAME,
                    handoffs=[agent_sug, agent_reg]
                )

                # 新しいユーザー入力を今までのやりとりに追加
                current_input = conversation_items + [{"role": "user", "content": user_input}]

                # 親エージェント「1次受付」に回答要求
                result = await Runner.run(agent_main, current_input)
                ret = result.final_output.strip() 
                
                # 今までのやりとりを抽出・保存（以降も会話のコンテキストとして使用）
                conversation_items = result.to_input_list()

            # async with ブロックを抜け、MCPサーバーとのMCPセッションが自動的に遮断される
                
            return ret                

        except Exception as e:
            logger.error(f"プロンプト処理中にエラー発生: {type(e).__name__}: {e}")
            return f"[MCP] エラーが発生しました: {str(e)}"

    try:
        return asyncio.run(async_process())
    except Exception as e:
        logger.error(f"asyncio実行エラー: {type(e).__name__}: {e}")
        return f"[MCP] 実行エラー: {str(e)}"