Python to Build ChatBot with Full Source Code For Beginners

  • Python-wechaty is a unified conversational RPA SDK for chatbot makers.
  • Developer at least use six lines code to create chatbot based on multi-ims,
  • Eg: WeChat, WeChat official account, dingtalk, lark, whatsapp, giter, and so on .

There are Chinese documents and English documents for developers to create their own chatbots.


pip install -r projects/chatbot/requirements.txt


  • wechaty

Run the script:

  • Open on device using a python IDE
  • Run the script
python projects/chatbot/

Source Code:

"""example code for ding-dong-bot with oop style"""
from typing import List, Optional, Union
import asyncio
from datetime import datetime
from wechaty_puppet import get_logger
from wechaty import (

logger = get_logger(__name__)

class MyBot(Wechaty):
    listen wechaty event with inherited functions, which is more friendly for
    oop developer

    def __init__(self) -> None:
        """initialization function
        self.login_user: Optional[Contact] = None

    async def on_ready(self, payload: EventReadyPayload) -> None:
        """listen for on-ready event"""'ready event %s...', payload)

    # pylint: disable=R0912,R0914,R0915
    async def on_message(self, msg: Message) -> None:
        listen for message event
        from_contact: Contact = msg.talker()
        text: str = msg.text()
        room: Optional[Room] =
        msg_type: MessageType = msg.type()
        file_box: Optional[FileBox] = None
        if text == 'ding':
            conversation: Union[
                Room, Contact] = from_contact if room is None else room
            await conversation.ready()
            await conversation.say('dong')
            file_box = FileBox.from_url(
            await conversation.say(file_box)

        elif msg_type == MessageType.MESSAGE_TYPE_IMAGE:
  'receving image file')
            # file_box: FileBox = await msg.to_file_box()
            image: Image = msg.to_image()

            hd_file_box: FileBox = await image.hd()
            await hd_file_box.to_file('./hd-image.jpg', overwrite=True)

            thumbnail_file_box: FileBox = await image.thumbnail()
            await thumbnail_file_box.to_file('./thumbnail-image.jpg', overwrite=True)
            artwork_file_box: FileBox = await image.artwork()
            await artwork_file_box.to_file('./artwork-image.jpg', overwrite=True)
            # reply the image
            await msg.say(hd_file_box)

        # pylint: disable=C0301
        elif msg_type in [MessageType.MESSAGE_TYPE_AUDIO, MessageType.MESSAGE_TYPE_ATTACHMENT, MessageType.MESSAGE_TYPE_VIDEO]:
  'receving file ...')
            file_box = await msg.to_file_box()
            if file_box:
                await file_box.to_file(

        elif msg_type == MessageType.MESSAGE_TYPE_MINI_PROGRAM:
  'receving mini-program ...')
            mini_program: Optional[MiniProgram] = await msg.to_mini_program()
            if mini_program:
                await msg.say(mini_program)

        elif text == 'get room members' and room:
  'get room members ...')
            room_members: List[Contact] = await room.member_list()
            names: List[str] = [
       for room_member in room_members]
            await msg.say('\n'.join(names))

        elif text.startswith('remove room member:'):
  'remove room member:')
            if not room:
                await msg.say('this is not room zone')

            room_member_name = text[len('remove room member:') + 1:]

            room_member: Optional[Contact] = await room.member(
            if room_member:
                if self.login_user and self.login_user.contact_id in room.payload.admin_ids:
                    await room.delete(room_member)
                    await msg.say('登录用户不是该群管理员...')

                await msg.say(f'can not fine room member by name<{room_member_name}>')
        elif text.startswith('get room topic'):
  'get room topic')
            if room:
                topic: Optional[str] = await room.topic()
                if topic:
                    await msg.say(topic)

        elif text.startswith('rename room topic:'):
  'rename room topic ...')
            if room:
                new_topic = text[len('rename room topic:') + 1:]
                await msg.say(new_topic)
        elif text.startswith('add new friend:'):
  'add new friendship ...')
            identity_info = text[len('add new friend:'):]
            weixin_contact: Optional[Contact] = await
            phone_contact: Optional[Contact] = await
            contact: Optional[Contact] = weixin_contact or phone_contact
            if contact:
                await self.Friendship.add(contact, 'hello world ...')

        elif text.startswith('at me'):
            if room:
                talker = msg.talker()
                await room.say('hello', mention_ids=[talker.contact_id])

        elif text.startswith('my alias'):
            talker = msg.talker()
            alias = await talker.alias()
            await msg.say('your alias is:' + (alias or ''))

        elif text.startswith('set alias:'):
            talker = msg.talker()
            new_alias = text[len('set alias:'):]

            # set your new alias
            alias = await talker.alias(new_alias)
            # get your new alias
            alias = await talker.alias()
            await msg.say('your new alias is:' + (alias or ''))

        elif text.startswith('find friends:'):
            friend_name: str = text[len('find friends:'):]
            friend = await self.Contact.find(friend_name)
            if friend:
      'find only one friend <%s>', friend)

            friends: List[Contact] = await self.Contact.find_all(friend_name)

  'find friend<%d>', len(friends))


        if msg.type() == MessageType.MESSAGE_TYPE_UNSPECIFIED:
            talker = msg.talker()
            assert isinstance(talker, Contact)

    async def on_login(self, contact: Contact) -> None:
        """login event
            contact (Contact): the account logined
        """'Contact<%s> has logined ...', contact)
        self.login_user = contact

    async def on_friendship(self, friendship: Friendship) -> None:
        """when receive a new friendship application, or accept a new friendship
            friendship (Friendship): contains the status and friendship info,
                eg: hello text, friend contact object
        # 1. receive a new friendship from someone
        if friendship.type() == FriendshipType.FRIENDSHIP_TYPE_RECEIVE:
            hello_text: str = friendship.hello()

            # accept friendship when there is a keyword in hello text
            if 'wechaty' in hello_text.lower():
                await friendship.accept()

        # 2. you have a new friend to your contact list
        elif friendship.type() == FriendshipType.FRIENDSHIP_TYPE_CONFIRM:
            # 2.1 invite the user to wechaty group
            # find the topic of room which contains Wechaty keyword
            wechaty_rooms: List[Room] = await self.Room.find_all('Wechaty')

            # 2.2 find the suitable room
            for wechaty_room in wechaty_rooms:
                members: List[Contact] = await wechaty_room.member_list()
                if len(members) < MAX_ROOM_MEMBER_COUNT:
                    contact: Contact =
                    await wechaty_room.add(contact)

    async def on_room_join(self, room: Room, invitees: List[Contact],
                           inviter: Contact, date: datetime) -> None:
        """on_room_join when there are new contacts to the room
                room (Room): the room instance
                invitees (List[Contact]): the new contacts to the room
                inviter (Contact): the inviter who share qrcode or manual invite someone
                date (datetime): the datetime to join the room
        # 1. say something to welcome the new arrivals
        names: List[str] = []
        for invitee in invitees:
            await invitee.ready()

        await room.say(f'welcome {",".join(names)} to the wechaty group !')

async def main() -> None:
    bot = MyBot()
    await bot.start() language: HTML, XML (xml)

import asyncio
from wechaty import Wechaty, Message

async def on_message(msg: Message):
    if msg.text() == 'ding':
        await msg.say('dong')
async def main():
    bot = Wechaty()
    bot.on('message', on_message)
    await bot.start() language: JavaScript (javascript)

Leave a Comment