Agents

import asyncio

# We'll use pydantic to automate the creation of our input and output schemas.
from meshagent.agents.pydantic import PydanticAgent, BaseModel
from meshagent.agents import AgentCallContext, connect_development_agent

# define the structure of the input of our agent
class Input(BaseModel):
    model_config = dict(extra="forbid")
    a: int
    b: int

# define the structure of the output of our agent
class Output(BaseModel):
    model_config = dict(extra="forbid")
    result: int

class Adder(PydanticAgent):
    def __init__(self):
        super().__init__(
            name="samples.adder",
            title="sample adder",
            description="an agent that adds two numbers",
            input_model=Input,
            output_model=Output,
        )

    # Our agent will receive content matching the input format, and must return content matching the output format
    async def ask_model(self, context: AgentCallContext, arguments: Input) -> Output:        
        result = arguments.a + arguments.b
        
        # let's log a message, it will show up in the developer console
        context.room.developer.log_nowait(type="sample", data={"message" : "hello world" })
        
        return Output(result=result)

async def main():
    room_name = "examples"

    # start our agent in developer mode, it will connect to the room and be available immediately from the admin console UI    
    await connect_development_agent(room_name=room_name, agent=Adder())
            


if __name__ == '__main__':
    
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    asyncio.get_event_loop().run_until_complete(main())
    

import asyncio

# We'll use pydantic to automate the creation of our input and output schemas.
from meshagent.agents.pydantic import PydanticWriter, BaseModel, WriterContext
from meshagent.agents import connect_development_agent

# define the structure of the input of our agent
class Input(BaseModel):
    model_config = dict(extra="forbid")
    # input must contain a path property, the base writer will write to the document at the specified path
    path: str
    content: str

# define the structure of the output of our agent
class Output(BaseModel):
    model_config = dict(extra="forbid")
    content: str

class SampleWriter(PydanticWriter):
    def __init__(self):
        super().__init__(
            name="samples.writer",
            title="sample writer",
            description="a agent that writes to a document",
            input_model=Input,
            output_model=Output,
        )

    # Our writer agent will receive a copy of the document. Any changes we make to the document will be automatically
    # saved and synchronized to other participants in the room.
    async def write_model(self, writer_context: WriterContext, arguments: Input) -> Output:
        print(f"writing content {arguments.content}")
        writer_context.document.root.append_child(tag_name="body", attributes={"text":arguments.content})
        return Output(content="success")

async def main():
    room_name = "examples"

    # start our agent in developer mode, it will connect to the room and be available immediately from the admin console UI    
    await connect_development_agent(room_name=room_name, agent=SampleWriter())
            

if __name__ == '__main__':
    
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    asyncio.get_event_loop().run_until_complete(main())

import asyncio
import os
from meshagent.api import RoomClient, ParticipantToken, WebSocketClientProtocol, websocket_room_url

async def main():

    room_name = "quickstart"

    # a participant token says who this participant is and what they are allowed to do.
    # Make sure your API secret in the MESHAGENT_SECRET environment variable. 
    # This secret will be used to sign the token so that the server knows that the token is valid.
    token = ParticipantToken(
        name="my user",
        project_id = os.getenv("MESHAGENT_PROJECT_ID"),
        api_key_id = os.getenv("MESHAGENT_KEY_ID")
    )

    token.add_room_grant(room_name=room_name)

    # connect to the room, it will automatically be created if it does not exist yet. 
    # We'll establish a connection to the server using a websocket.
    url = websocket_room_url(room_name=room_name)

    async with RoomClient(
        protocol=WebSocketClientProtocol(
            url = url,
            token = token.to_jwt(token=os.getenv("MESHAGENT_SECRET"))
        )) as room:            

            path = "sample-agent.document"
            

            # We can ask an agent to perform some work in the document, in this case, we'll use the built in
            # document_writer agent to write content to our document.
            # The document will automatically be created if it does not already exist.

            await room.agents.ask(
                agent="meshagent.document-writer",
                arguments={
                    "path": path,
                    "prompt":"write a paragraph about ai and how agents are shaping the future"
                }
            )
            print(f"Take a look at it at https://studio.meshagent.com/projects/{os.getenv("MESHAGENT_PROJECT_ID")}/rooms/{room_name}?p={path}")


if __name__ == '__main__':
    
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    asyncio.get_event_loop().run_until_complete(main())