MetaGPT Custom Roles and Actions: Build Your Own Software Team

Q: How do roles communicate in MetaGPT?

Roles communicate via a shared message bus. When a role publishes a Message with causeby=SomeAction, any role watching SomeAction receives it. This is a publish-subscribe pattern — roles don't call each other directly.

Q: How do I control which role handles which messages?

Use watch([ActionClass]) to filter incoming messages. A role only acts on messages caused by actions in its watch list. Use sendto in the Message to target specific roles directly.

Q: Can roles use external tools (web search, code execution)?

Yes. Define tool-calling logic inside Action's run() method. You can call any Python code, make HTTP requests, run subprocesses, or use frameworks like LangChain inside an action.

Q: How do I debug the conversation between roles?

Set logger.setLevel("DEBUG") to see all messages being passed between roles. The Team.run() output also shows each role's actions and outputs.

MetaGPT’s Role/Action Architecture

MetaGPT structures agents as Roles that perform Actions. The built-in software company has predefined roles (ProductManager, Architect, Engineer, QA), but you can define your own for any domain.

The key abstractions:

Action — a single unit of work (write a document, generate code, review output)
Role — an agent that executes actions in response to messages
Team — a group of roles that collaborate

This architecture models any professional team: editorial staff, financial analysts, legal reviewers, research assistants.

Creating a Custom Action

from metagpt.actions import Action
from metagpt.schema import Message

class WriteResearchReport(Action):
    """Action that researches a topic and writes a structured report."""

    PROMPT_TEMPLATE: str = """
You are a research analyst. Write a comprehensive report on the following topic.

Topic: {topic}
Focus areas: {focus_areas}

Structure the report with:
1. Executive Summary (2-3 sentences)
2. Background and Context
3. Key Findings (3-5 bullet points with evidence)
4. Analysis and Implications
5. Recommendations
6. Sources (list as bullet points)

Be specific, data-driven, and professional. Minimum 500 words.
"""

    name: str = "WriteResearchReport"

    async def run(self, topic: str, focus_areas: str = "general overview") -> str:
        prompt = self.PROMPT_TEMPLATE.format(
            topic=topic,
            focus_areas=focus_areas,
        )
        rsp = await self._aask(prompt)
        return rsp

Creating a Custom Role

from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger

class ResearchAnalyst(Role):
    """A role that researches topics and produces detailed reports."""

    name: str = "Alex"
    profile: str = "Research Analyst"
    goal: str = "Research topics thoroughly and produce clear, accurate reports"
    constraints: str = "Base all claims on evidence. Flag uncertainties clearly."

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Register the actions this role can perform
        self.set_actions([WriteResearchReport])
        # React to incoming research requests
        self._watch([UserRequirement])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: preparing to write research report")

        # Get the latest message from memory
        todo = self.rc.todo  # the action to execute
        msg = self.get_memories(k=1)[0]  # get last message

        # Run the action
        code_text = await WriteResearchReport().run(topic=msg.content)

        msg = Message(
            content=code_text,
            role=self.profile,
            cause_by=type(todo),
        )
        return msg

Multi-Role Example: Editorial Team

Build a complete editorial workflow with custom roles:

import asyncio
from metagpt.actions import Action, UserRequirement
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team

# ── Actions ──────────────────────────────────────────────────────

class ResearchTopic(Action):
    name: str = "ResearchTopic"

    async def run(self, topic: str) -> str:
        prompt = f"""Research this topic and provide key facts, statistics, and sources:
Topic: {topic}

Provide:
- 5 key facts
- 2-3 statistics with sources
- Main controversies or debates
- Expert opinions
"""
        return await self._aask(prompt)

class WriteArticle(Action):
    name: str = "WriteArticle"

    async def run(self, research: str, topic: str) -> str:
        prompt = f"""Write a 600-word article for a developer blog based on this research.

Topic: {topic}
Research findings:
{research}

Requirements:
- Engaging introduction
- 3 main sections with headers
- Include specific examples and code snippets if relevant
- Practical takeaways
- Short conclusion
"""
        return await self._aask(prompt)

class EditArticle(Action):
    name: str = "EditArticle"

    async def run(self, article: str) -> str:
        prompt = f"""Edit this article for quality, clarity, and accuracy.

Article:
{article}

Improvements to make:
- Fix grammar and style issues
- Improve sentence flow
- Ensure technical accuracy
- Strengthen the introduction and conclusion
- Return the complete edited article
"""
        return await self._aask(prompt)

# ── Roles ─────────────────────────────────────────────────────────

class Researcher(Role):
    name: str = "Sam"
    profile: str = "Research Specialist"
    goal: str = "Research topics thoroughly with facts and citations"
    constraints: str = "Only use verifiable information"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ResearchTopic])
        self._watch([UserRequirement])

    async def _act(self) -> Message:
        msg = self.get_memories(k=1)[0]
        research = await ResearchTopic().run(topic=msg.content)
        return Message(content=research, role=self.profile, cause_by=ResearchTopic)

class Writer(Role):
    name: str = "Jordan"
    profile: str = "Content Writer"
    goal: str = "Write clear, engaging articles from research briefs"
    constraints: str = "Follow the house style guide"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteArticle])
        self._watch([ResearchTopic])  # trigger after research is done

    async def _act(self) -> Message:
        msgs = self.get_memories(k=2)
        research = msgs[-1].content if msgs else ""
        original_request = msgs[0].content if len(msgs) > 1 else "AI agents"

        article = await WriteArticle().run(research=research, topic=original_request)
        return Message(content=article, role=self.profile, cause_by=WriteArticle)

class Editor(Role):
    name: str = "Morgan"
    profile: str = "Senior Editor"
    goal: str = "Polish articles to publication quality"
    constraints: str = "Maintain author voice, only improve quality"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([EditArticle])
        self._watch([WriteArticle])  # trigger after writing is done

    async def _act(self) -> Message:
        msgs = self.get_memories(k=1)
        article = msgs[0].content if msgs else ""

        edited = await EditArticle().run(article=article)
        return Message(content=edited, role=self.profile, cause_by=EditArticle)

# ── Run the Team ──────────────────────────────────────────────────

async def main():
    team = Team()
    team.hire([
        Researcher(),
        Writer(),
        Editor(),
    ])

    # Investment controls how many LLM calls are made
    team.invest(5.0)
    team.run_project("Write about the impact of persistent memory on AI agent performance")

    await team.run(n_round=5)

asyncio.run(main())

Role Communication Patterns

Watch Pattern

Roles respond to specific action types:

class QualityChecker(Role):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([RunQAChecks])
        # Triggers after ANY writing action completes
        self._watch([WriteArticle, WriteResearchReport, WriteCode])

Direct Message Pattern

Send a targeted message to a specific role:

from metagpt.schema import Message

# From within an action or orchestration code
msg = Message(
    content="Please review this code for security issues",
    role="SecurityReviewer",
    send_to={"SecurityReviewer"},  # targeted delivery
    cause_by=WriteCode,
)

Adding Memory to Custom Roles

from metagpt.memory import Memory
from metagpt.schema import Message

class StatefulResearcher(Role):
    name: str = "Dr. Chen"
    profile: str = "Senior Researcher"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([ResearchTopic])
        self._watch([UserRequirement])

        # Custom memory storage
        self._researched_topics: list[str] = []

    async def _act(self) -> Message:
        msg = self.get_memories(k=1)[0]
        topic = msg.content

        # Check if we've researched this before
        if topic in self._researched_topics:
            # Use existing memory
            existing = self.get_memories(k=5)
            related = [m for m in existing if topic.lower() in m.content.lower()]
            if related:
                return Message(
                    content=f"Already researched: {related[-1].content[:500]}",
                    role=self.profile,
                    cause_by=ResearchTopic,
                )

        # New topic — research it
        result = await ResearchTopic().run(topic=topic)
        self._researched_topics.append(topic)

        return Message(content=result, role=self.profile, cause_by=ResearchTopic)

Frequently Asked Questions

How do roles communicate in MetaGPT?

Roles communicate via a shared message bus. When a role publishes a Message with cause_by=SomeAction, any role watching SomeAction receives it. This is a publish-subscribe pattern — roles don’t call each other directly.

Can I mix custom and built-in roles?

Yes. You can hire both custom and built-in roles in the same Team:

from metagpt.roles import Engineer
team.hire([Researcher(), Writer(), Engineer()])

How do I control which role handles which messages?

Use _watch([ActionClass]) to filter incoming messages. A role only acts on messages caused by actions in its watch list. Use send_to in the Message to target specific roles directly.

Can roles use external tools (web search, code execution)?

Yes. Define tool-calling logic inside Action’s run() method. You can call any Python code, make HTTP requests, run subprocesses, or use frameworks like LangChain inside an action.

How do I debug the conversation between roles?

Set logger.setLevel("DEBUG") to see all messages being passed between roles. The Team.run() output also shows each role’s actions and outputs.

Next Steps

MetaGPT Use Cases and Examples — See custom role patterns in practice
MetaGPT Data Interpreter — Use MetaGPT’s specialized data agent
CrewAI Multi-Agent Workflows — Compare with CrewAI’s role-based approach