[#26] 예시용 MCP 서버 생성 및 실행 스크립트 생성

.env.example

@@ -6,17 +6,24 @@
 # LangSmith is a tool for monitoring and debugging 
 LANGSMITH_PROJECT=pseudo-entertainment-company  # Project name to be used by LangSmith.
 LANGSMITH_API_KEY=lsv2....  # LangSmith API key (must be replaced with real key)
+LANGGRAPH_PORT=2024
 
-# Depending on the configuration you choose, you will need the following environment variables.
-
-## LLM API Keys:
+# LLM API Keys:
 # OpenAI API Key - required to use the GPT model.
 # You can get it from the OpenAI website (https://platform.openai.com/).
 OPENAI_API_KEY=sk...
 
-## Groq
+
+# Google API Key (예시에는 없지만, 필요하다면 주석 추가)
+GOOGLE_API_KEY=...
+
+# Upstage API Key (예시에는 없지만, 필요하다면 주석 추가)
+UPSTAGE_API_KEY=...
+
+# News API Key - required to use the news API.
+# You can get it from the News API website (https://newsapi.org/).
+NEWS_API_KEY=...
+
 # Groq API Key - used to access Groq LLMs such as Mixtral or LLaMA models.
 # Sign up and get your key from https://console.groq.com/keys
 GROQ_API_KEY=grq...
-
-# Others...

agents/text/.env.sample

@@ -0,0 +1,4 @@
+# MCP 서버
+MCP_NEWS_HOST=0.0.0.0
+MCP_NEWS_PORT=8100
+MCP_NEWS_TRANSPORT=stdio

agents/text/README.md

@@ -2,11 +2,13 @@
 
 ## 개요
 
-이 모듈은 Pseudo Entertainment Company의 텍스트 기반 콘텐츠 생성을 담당하는 LangGraph Workflow입니다. 다양한 유형의 텍스트 콘텐츠를 생성하기 위한 주요 노드와 Workflow를 제공합니다.
+이 모듈은 Act 1: Entertainment의 텍스트 기반 콘텐츠 생성을 담당하는 LangGraph Workflow입니다. 다양한 유형의 텍스트 콘텐츠를 생성하기 위한 주요 노드와 Workflow를 제공합니다.
 
 ## 주요 노드
 
-<!-- 노드에 대한 설명을 추가해주세요. -->
+- `PersonaExtractionNode`: 콘텐츠 종류에 적합한 페르소나를 추출하는 노드
+- `GenTextNode`: 추출된 페르소나를 바탕으로 인스타그램 포스트에 적합한 텍스트를 생성하는 노드
+- `TopicFromNewsNode`: 주어진 키워드에 대한 뉴스 기사를 스크래핑하는 노드
 
 ## 구조
 
@@ -26,6 +28,35 @@ text/
 └── workflow.py        # Text Agent의 Workflow들 정의
 ```
 
+## 실행 방법
+
+1. `langgraph.json` 파일을 Text Agent에 맞춰 설정해주세요.
+
+```json
+{
+  "dependencies": ["./agents/text"],
+  "graphs": {
+    "text": "./agents/text/workflow.py:text_workflow"
+  },
+  "env": ".env"
+}
+```
+
+2. `agents/text/text_agent.sh` 파일에 실행 권한을 부여하고 실행하세요.
+
+```bash
+$ chmod +x agents/text/text_agent.sh
+$ agents/text/text_agent.sh
+```
+
+> 로컬 포트에 이미 연결된 프로세스가 존재하는 경우, 다음과 같은 문구가 CLI 창에 출력됩니다.
+>
+> ```bash
+> 포트 <PORT> 사용 중인 프로세스<PID>가 있습니다. 종료하시겠습니까? (y/N):
+> ```
+>
+> PID로 해당 프로세스를 확인하고 실행해주세요.
+
 ## 사용 방법
 
 텍스트 Workflow는 다음과 같이 사용할 수 있습니다:
@@ -56,4 +87,4 @@ result = text_workflow().invoke(initial_state)
 
 ## 라이센스
 
-이 모듈은 Pseudo Group의 Pseudo Entertainment Company의 내부 프로젝트로, 그룹 정책에 따른 라이센스가 적용됩니다.
+이 모듈은 Proact0 의 Act 1: Entertainment 의 내부 프로젝트로, 그룹 정책에 따른 라이센스가 적용됩니다.

agents/text/mcp/mcp_client.py

@@ -0,0 +1,40 @@
+import os
+
+from langchain_mcp_adapters.tools import load_mcp_tools
+from langgraph.prebuilt import create_react_agent
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from agents.text.modules.models import get_openai_model
+
+
+async def scrape_news(scraping_query: str) -> str:
+    """
+    비동기적으로 뉴스 스크래핑 MCP 서버에 요청을 보내고, 응답을 반환합니다.
+    """
+
+    server_params = StdioServerParameters(
+        command="uv",
+        args=["--directory", "agents/text/mcp/", "run", "mcp_news_server.py"],
+        env=os.environ,
+    )
+
+    async with stdio_client(server=server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+
+            tools = await load_mcp_tools(session)
+            print("사용 가능한 도구:", tools)
+
+            graph = create_react_agent(model=get_openai_model(), tools=tools)
+
+            response = await graph.ainvoke({"messages": scraping_query})
+
+            if "messages" in response:
+                messages = response["messages"]
+                for message in messages:
+                    if hasattr(message, "content") and message.content:
+                        last_content = message.content
+                        break
+
+            return last_content

agents/text/mcp/mcp_news_server.py

@@ -0,0 +1,48 @@
+import os
+from datetime import date, timedelta
+
+from mcp.server.fastmcp import FastMCP
+from newsapi import NewsApiClient
+
+NEWS_API_KEY = os.getenv("NEWS_API_KEY")
+MCP_NEWS_HOST = os.getenv("MCP_NEWS_HOST", "0.0.0.0")
+MCP_NEWS_PORT = int(os.getenv("MCP_NEWS_PORT", 8100))
+MCP_NEWS_TRANSPORT = os.getenv("MCP_NEWS_TRANSPORT", "stdio")
+
+news_mcp = FastMCP(
+    name="news",
+    instructions=(
+        "Act as a news-scraping assistant that, given today's date, finds today's news for Instagram text content."
+    ),
+    host=MCP_NEWS_HOST,
+    port=MCP_NEWS_PORT,
+)
+news_api = NewsApiClient(api_key=NEWS_API_KEY)
+
+
+@news_mcp.tool()
+async def find_news(keywords: str) -> list[dict]:
+    """
+    키워드를 기반으로 뉴스 기사를 찾습니다.
+
+    Args:
+        keywords (str): 검색할 뉴스 기사 키워드
+
+    Returns:
+        list[dict]: 뉴스 기사 목록 (최대 5개)
+    """
+    from_date = (date.today() - timedelta(days=7)).isoformat()  # 7일 전까지만 기사 검색
+
+    news: list[dict] = news_api.get_everything(
+        q=keywords,
+        from_param=from_date,
+        sort_by="popularity",  # 인기도 순으로 정렬
+    )["articles"]
+
+    return news[:5]  # 상위 5개 기사만 반환
+
+
+if __name__ == "__main__":
+    print(f"news MCP server is running on {MCP_NEWS_HOST}:{MCP_NEWS_PORT}")
+
+    news_mcp.run(transport=MCP_NEWS_TRANSPORT)

agents/text/modules/chains.py

@@ -5,19 +5,23 @@
 
 """
 
-from langchain.schema.runnable import (
+
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.runnables import (
     RunnableLambda,
     RunnableMap,
     RunnablePassthrough,
     RunnableSerializable,
 )
-from langchain_core.output_parsers import StrOutputParser
 
+from agents.text.mcp.mcp_client import scrape_news
 from agents.text.modules.models import get_groq_model, get_openai_model
 from agents.text.modules.persona import PERSONA
 from agents.text.modules.prompts import (
     get_extraction_prompt,
     get_instagram_text_prompt,
+    get_news_scraping_query_prompt,
+    get_topic_from_news_prompt,
     get_persona_match_prompt,
 )
 
@@ -58,6 +62,35 @@ def set_extraction_chain() -> RunnableSerializable:
     )
 
 
+def set_topic_generation_news_chain() -> RunnableSerializable:
+    """
+    뉴스로부터 텍스트 콘텐츠 주제를 추출하는 LangChain 체인을 생성합니다.
+    """
+    news_scraping_query_prompt = get_news_scraping_query_prompt()
+    model = get_openai_model()
+
+    # 문자열 입력을 딕셔너리로 변환
+    input_transformer = RunnableLambda(lambda x: {"content_topic": x})
+
+    news_scraping_query_chain = (
+        input_transformer
+        | RunnablePassthrough.assign(content_topic=lambda x: x["content_topic"])
+        | news_scraping_query_prompt
+        | model
+        | StrOutputParser()  # 결과를 문자열로 변환
+    )
+
+    return (
+        news_scraping_query_chain
+        | RunnableLambda(scrape_news)
+        | RunnableLambda(lambda x: {"news_article": x})  # 반환값을 딕셔너리로 변환
+        | RunnablePassthrough.assign(persona_details=lambda x: PERSONA)
+        | get_topic_from_news_prompt()
+        | model
+        | StrOutputParser()
+    )
+
+
 def set_instagram_text_chain() -> RunnableSerializable:
     """
     인스타그램 텍스트 생성에 사용할 LangChain 체인을 생성합니다.

agents/text/modules/nodes.py

@@ -4,10 +4,13 @@
 해당 클래스 모듈은 각각 노드 클래스가 BaseNode를 상속받아 노드 클래스를 구현하는 모듈입니다.
 """
 
+import asyncio
+
 from agents.base_node import BaseNode
 from agents.text.modules.chains import (
     set_extraction_chain,
     set_instagram_text_chain,
+    set_topic_generation_news_chain,
     set_text_content_check_chain,
 )
 from agents.text.modules.persona import PERSONA
@@ -68,6 +71,27 @@ def execute(self, state: TextState) -> dict:
         }
 
 
+class TopicFromNewsNode(BaseNode):
+    """
+    주어진 키워드에 대한 뉴스 기사를 스크래핑하는 노드
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.chain = set_topic_generation_news_chain()
+
+    def execute(self, state: TextState) -> dict:
+        """
+        주어진 키워드에 대한 뉴스 기사를 스크래핑하고, 결과를 응답으로 반환합니다.
+        """
+        try:
+            result = asyncio.run(self.chain.ainvoke(state["content_topic"]))
+            state["news"] = result
+            return {"response": result}
+        except Exception as e:
+            return {"response": f"뉴스 검색 중 오류가 발생했습니다: {str(e)}"}
+
+
 class TextContentCheckNode(BaseNode):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)

agents/text/modules/prompts.py

@@ -100,6 +100,48 @@ def get_instagram_text_prompt():
     )
 
 
+def get_news_scraping_query_prompt() -> PromptTemplate:
+    """
+    뉴스 스크래핑을 위한 프롬프트 템플릿을 생성합니다.
+    """
+    prompt_template = """You are **News Summary Agent** that search news articles based on given content topic:
+{content_topic}
+
+You have to summarize news articles based on given content topic for instagram post.
+
+News Summary: """
+
+    return PromptTemplate(
+        template=prompt_template,
+        input_variables=["content_topic"],
+    )
+
+
+def get_topic_from_news_prompt():
+    """
+    뉴스로부터 인스타그램 포스트 주제를 추출하는 프롬프트를 생성합니다.
+    """
+    prompt_template = """
+You are an assistant responsible for extracting a concise and engaging topic from a given news article.
+You have to generate a topic that is relevant to the news article and is suitable for an instagram post.
+Also topic must be aligned with given persona.
+
+Rules:
+- The topic must be short and clear.
+- It must be relevant to the news article.
+- Output only the topic—no extra text.
+
+news article: {news_article}
+persona: {persona_details}
+
+topic: """
+
+    return PromptTemplate(
+        template=prompt_template,
+        input_variables=["news_article", "persona_details"],
+    )
+
+
 def get_persona_match_prompt() -> PromptTemplate:
     """
     Returns a prompt template to evaluate if a given text aligns with a provided persona.

agents/text/modules/state.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Annotated, TypedDict
+from typing import Annotated, Optional, TypedDict
 
 from langgraph.graph.message import add_messages
 
@@ -18,9 +18,10 @@ class TextState(TypedDict):
     content_topic: str  # 콘텐츠의 주제 (예: "여름 휴가", "음식 리뷰")
     content_type: str  # 콘텐츠의 유형 (예: "블로그 글", "소셜 미디어 포스트")
     query: str  # 사용자 쿼리 또는 요청사항
-    persona_extracted: str  # 추출된 페르소나 전문
-    instagram_text: str  # 생성된 인스타그램 텍스트
-    response: Annotated[
-        list, add_messages
-    ]  # 응답 메시지 목록 (add_messages로 주석되어 메시지 추가 기능 제공)
+    persona_extracted: Optional[str] = None  # 추출된 페르소나 전문
+    news: Optional[str] = None  # 뉴스 기사
+    instagram_text: Optional[str] = None  # 생성된 인스타그램 텍스트
+    response: Optional[Annotated[list, add_messages]] = (
+        None  # 응답 메시지 목록 (add_messages로 주석되어 메시지 추가 기능 제공)
+    )
     text_content_checker_result: (dict)  # 텍스트 컨텐츠 검사 결과 전체를 담는 구조화된 필드

agents/text/pyproject.toml

@@ -10,6 +10,12 @@ description = "텍스트 기반 콘텐츠 생성을 위한 LangGraph Workflow 
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "langchain-mcp-adapters>=0.0.9",
     "langchain-groq>=0.3.2",
     "langchain-openai>=0.3.12",
+    "mcp>=1.6.0",
+    "newsapi-python>=0.2.7",
 ]
+
+[tool.setuptools]
+packages = ["modules", "mcp"]