Function bodies 807 total

AgentMessage.to_dict method · python · L71-L80 (10 LOC)

fabric_agent/agents/base.py

    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "from_agent": self.from_agent,
            "to_agent": self.to_agent,
            "message_type": self.message_type.value,
            "content": self.content,
            "timestamp": self.timestamp,
            "correlation_id": self.correlation_id,
        }

ToolDefinition.to_llm_format method · python · L91-L100 (10 LOC)

fabric_agent/agents/base.py

    def to_llm_format(self) -> Dict[str, Any]:
        """Convert to format expected by LLM APIs."""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": {
                "type": "object",
                "properties": self.parameters,
            }
        }

AgentResult.to_dict method · python · L133-L143 (11 LOC)

fabric_agent/agents/base.py

    def to_dict(self) -> Dict[str, Any]:
        return {
            "agent_name": self.agent_name,
            "task": self.task,
            "status": self.status.value,
            "result": self.result,
            "thoughts": [vars(t) for t in self.thoughts],
            "tool_calls": [vars(tc) for tc in self.tool_calls],
            "error": self.error,
            "execution_time_ms": self.execution_time_ms,
        }

LLMClient.complete method · python · L209-L216 (8 LOC)

fabric_agent/agents/base.py

    async def complete(
        self,
        messages: List[Dict[str, str]],
        tools: Optional[List[Dict]] = None,
        system_prompt: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Send completion request to LLM."""
        pass

ClaudeLLMClient.complete method · python · L227-L265 (39 LOC)

fabric_agent/agents/base.py

    async def complete(
        self,
        messages: List[Dict[str, str]],
        tools: Optional[List[Dict]] = None,
        system_prompt: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Send completion request to Claude."""
        try:
            import anthropic
            
            if self._client is None:
                self._client = anthropic.AsyncAnthropic(api_key=self.api_key)
            
            kwargs = {
                "model": self.model,
                "max_tokens": 4096,
                "messages": messages,
            }
            
            if system_prompt:
                kwargs["system"] = system_prompt
            
            if tools:
                kwargs["tools"] = tools
            
            response = await self._client.messages.create(**kwargs)
            
            return {
                "content": response.content,
                "stop_reason": response.stop_reason,
                "usage": {
                    "

MockLLMClient.complete method · python · L275-L292 (18 LOC)

fabric_agent/agents/base.py

    async def complete(
        self,
        messages: List[Dict[str, str]],
        tools: Optional[List[Dict]] = None,
        system_prompt: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Return mock response."""
        if self.responses and self.call_count < len(self.responses):
            response = self.responses[self.call_count]
            self.call_count += 1
            return response
        
        # Default mock response
        return {
            "content": [{"type": "text", "text": "Mock response"}],
            "stop_reason": "end_turn",
            "usage": {"input_tokens": 100, "output_tokens": 50}
        }

BaseAgent.__init__ method · python · L316-L343 (28 LOC)

fabric_agent/agents/base.py

    def __init__(
        self,
        name: str,
        role: AgentRole,
        llm: LLMClient,
        memory: SharedMemory,
        max_iterations: int = 10,
    ):
        """
        Initialize the agent.
        
        Args:
            name: Unique agent name
            role: Agent's role
            llm: LLM client for reasoning
            memory: Shared memory
            max_iterations: Max think-act cycles
        """
        self.name = name
        self.role = role
        self.llm = llm
        self.memory = memory
        self.max_iterations = max_iterations
        self._tools: Dict[str, ToolDefinition] = {}
        
        # Register tools
        for tool in self.get_tools():
            self._tools[tool.name] = tool

About: code-quality intelligence by Repobility · https://repobility.com

BaseAgent.run method · python · L355-L516 (162 LOC)

fabric_agent/agents/base.py

    async def run(
        self,
        task: str,
        context: Optional[Dict[str, Any]] = None,
    ) -> AgentResult:
        """
        Execute the agent on a task.
        
        Args:
            task: The task to perform
            context: Additional context
        
        Returns:
            AgentResult with outcome
        """
        import time
        start_time = time.time()
        
        logger.info(f"[{self.name}] Starting task: {task[:100]}...")
        
        thoughts: List[AgentThought] = []
        tool_calls: List[ToolCall] = []
        
        # Build initial messages
        messages = [
            {"role": "user", "content": self._build_task_prompt(task, context)}
        ]
        
        # Get tools in LLM format
        llm_tools = [t.to_llm_format() for t in self._tools.values()] if self._tools else None
        
        try:
            for iteration in range(self.max_iterations):
                # Call LLM
                response = await

BaseAgent._call_tool method · python · L518-L529 (12 LOC)

fabric_agent/agents/base.py

    async def _call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """Execute a tool."""
        tool = self._tools.get(tool_name)
        if not tool:
            raise ValueError(f"Unknown tool: {tool_name}")
        
        # Call handler (may be sync or async)
        result = tool.handler(**arguments)
        if asyncio.iscoroutine(result):
            result = await result
        
        return result

BaseAgent._build_task_prompt method · python · L531-L538 (8 LOC)

fabric_agent/agents/base.py

    def _build_task_prompt(self, task: str, context: Optional[Dict]) -> str:
        """Build the task prompt with context."""
        prompt = f"Task: {task}"
        
        if context:
            prompt += f"\n\nContext:\n{json.dumps(context, indent=2)}"
        
        return prompt

BaseAgent.send_message method · python · L544-L554 (11 LOC)

fabric_agent/agents/base.py

    def send_message(self, to_agent: str, content: Dict[str, Any]) -> AgentMessage:
        """Send a message to another agent."""
        message = AgentMessage(
            id=str(uuid4()),
            from_agent=self.name,
            to_agent=to_agent,
            message_type=MessageType.REQUEST,
            content=content,
        )
        self.memory.add_message(message)
        return message

SimpleAgent.__init__ method · python · L568-L593 (26 LOC)

fabric_agent/agents/base.py

    def __init__(
        self,
        name: str,
        role: AgentRole,
        memory: SharedMemory,
        tools: List[ToolDefinition],
    ):
        # IMPORTANT:
        # BaseAgent.__init__ registers tools by calling self.get_tools().
        # SimpleAgent.get_tools() returns self._simple_tools, so we must
        # set it *before* calling super().__init__.
        self._simple_tools = tools

        # Use mock LLM since we won't use it
        super().__init__(
            name=name,
            role=role,
            llm=MockLLMClient(),
            memory=memory,
            max_iterations=1,
        )

        # BaseAgent already registered tools via get_tools(), but keep this
        # to ensure tools are present even if BaseAgent behavior changes.
        for tool in tools:
            self._tools[tool.name] = tool

OrchestratorLLMAgent.__init__ method · python · L200-L236 (37 LOC)