Coverage for mindsdb / api / a2a / __init__.py: 0%

1# A2A specific imports 

2from mindsdb.api.a2a.common.types import ( 

3 AgentCard, 

4 AgentCapabilities, 

5 AgentSkill, 

6) 

7from mindsdb.api.a2a.common.server.server import A2AServer 

8from mindsdb.api.a2a.task_manager import AgentTaskManager 

9from mindsdb.api.a2a.agent import MindsDBAgent 

10from mindsdb.utilities.config import config 

11 

12 

13def get_a2a_app( 

14 project_name: str = "mindsdb", 

15): 

16 mindsdb_port = config.get("api", {}).get("http", {}).get("port", 47334) 

17 

18 # Prepare A2A artefacts (agent card & task-manager) 

19 capabilities = AgentCapabilities(streaming=True) 

20 skill = AgentSkill( 

21 id="mindsdb_query", 

22 name="MindsDB Query", 

23 description="Executes natural-language queries via MindsDB agents.", 

24 tags=["database", "mindsdb", "query", "analytics"], 

25 examples=[ 

26 "What trends exist in my sales data?", 

27 "Generate insights from the support tickets dataset.", 

28 ], 

29 inputModes=MindsDBAgent.SUPPORTED_CONTENT_TYPES, 

30 outputModes=MindsDBAgent.SUPPORTED_CONTENT_TYPES, 

31 ) 

32 

33 agent_card = AgentCard( 

34 name="MindsDB Agent Connector", 

35 description=(f"A2A connector that proxies requests to MindsDB agents in project '{project_name}'."), 

36 url=f"http://127.0.0.1:{mindsdb_port}/a2a/", 

37 version="1.0.0", 

38 defaultInputModes=MindsDBAgent.SUPPORTED_CONTENT_TYPES, 

39 defaultOutputModes=MindsDBAgent.SUPPORTED_CONTENT_TYPES, 

40 capabilities=capabilities, 

41 skills=[skill], 

42 ) 

43 

44 task_manager = AgentTaskManager( 

45 project_name=project_name, 

46 ) 

47 

48 server = A2AServer( 

49 agent_card=agent_card, 

50 task_manager=task_manager, 

51 ) 

52 return server.app